Apache Kafka용 Pulsar 어댑터

Apache Kafka용 Pulsar 어댑터

Pulsar는 현재 Apache Kafka Java 클라이언트 API로 작성된 애플리케이션을 위한 쉬운 선택지를 제공해요. 기존 Kafka 코드를 거의 그대로 두고 Pulsar로 옮길 수 있도록 Pulsar Kafka 래퍼(wrapper)를 제공하는데요. 다만 이 래퍼를 사용하기 전에 몇 가지 버전 규칙을 알아두는 게 좋아요.

출처: 문서

본문

note

Pulsar Kafka 래퍼 아티팩트(pulsar-client-kafka와 pulsar-client-kafka-original)는 별도의 apache/pulsar-adapters 리포지토리에 있으며, Maven Central에 게시된 마지막 릴리스 버전은 2.11.0이에요. 더 새로운 Pulsar 브로커(3.x / 4.x)를 실행하더라도 의존성은 2.11.0에 고정해 두세요 — 더 새로운 대응 아티팩트는 게시되지 않으니까요.

Pulsar Kafka 호환 래퍼 사용하기 (Use the Pulsar Kafka compatibility wrapper)

Pulsar Kafka 호환 래퍼를 사용하려면 다음 단계를 완료해요.

Step 1: 기존 애플리케이션에서 일반 Kafka 클라이언트 의존성을 Pulsar Kafka 래퍼로 교체해요. pom.xml의 다음 의존성을 제거하세요.

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>0.10.2.1</version>
</dependency>

Step 2: 그다음 Pulsar Kafka 래퍼에 대한 다음 의존성을 포함시키세요.

<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-client-kafka</artifactId>
  <version>2.11.0</version>
</dependency>

새 의존성을 사용하면 기존 코드가 변경 없이 동작해요. 설정을 조정해서 프로듀서와 컨슈머가 Kafka 대신 Pulsar 서비스를 가리키게 하고, 특정 Pulsar 토픽을 사용하도록 해야 해요.

기존 Kafka 클라이언트와 함께 Pulsar Kafka 래퍼 사용하기

Kafka에서 Pulsar로 마이그레이션할 때, 애플리케이션이 마이그레이션 기간 동안 원래 Kafka 클라이언트와 Pulsar Kafka 래퍼를 함께 사용할 수도 있어요. 이 경우 unshaded(언섀도) Pulsar Kafka 클라이언트 래퍼를 사용하는 것을 고려해야 해요.

<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-client-kafka-original</artifactId>
  <version>2.11.0</version>
</dependency>

이 의존성을 사용할 때는 프로듀서를 org.apache.kafka.clients.producer.KafkaProducer 대신 org.apache.kafka.clients.producer.PulsarKafkaProducer로, 컨슈머는 org.apache.kafka.clients.producer.PulsarKafkaConsumer로 생성해요.

프로듀서 예제 (Producer example)

// Topic needs to be a regular Pulsar topic
String topic = "persistent://public/default/my-topic";

Properties props = new Properties();
// Point to a Pulsar service
props.put("bootstrap.servers", "pulsar://localhost:6650");

props.put("key.serializer", IntegerSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());

Producer<Integer, String> producer = new KafkaProducer(props);

for (int i = 0; i < 10; i++) {
    producer.send(new ProducerRecord<Integer, String>(topic, i, "hello-" + i));
    log.info("Message {} sent successfully", i);
}

producer.close();

컨슈머 예제 (Consumer example)

String topic = "persistent://public/default/my-topic";

Properties props = new Properties();
// Point to a Pulsar service
props.put("bootstrap.servers", "pulsar://localhost:6650");
props.put("group.id", "my-subscription-name");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", IntegerDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

Consumer<Integer, String> consumer = new KafkaConsumer(props);
consumer.subscribe(Arrays.asList(topic));

while (true) {
    ConsumerRecords<Integer, String> records = consumer.poll(100);
    records.forEach(record -> {
        log.info("Received record: {}", record);
    });

    // Commit last offset
    consumer.commitSync();
}

전체 예제 (Complete Examples)

완전한 프로듀서와 컨슈머 예제는 여기에서 찾을 수 있어요.

호환성 매트릭스 (Compatibility matrix)

현재 Pulsar Kafka 래퍼는 Kafka API가 제공하는 대부분의 연산을 지원해요.

프로듀서 (Producer)

APIs:

Properties:

컨슈머 (Consumer)

다음 표는 컨슈머 API를 나열한 것이에요.

Properties:

Pulsar 설정 커스터마이징 (Customize Pulsar configurations)

Kafka 프로퍼티에서 직접 Pulsar 인증 제공자(auth provider)를 구성할 수 있어요.

Pulsar 클라이언트 프로퍼티 (Pulsar client properties)

Pulsar 프로듀서 프로퍼티 (Pulsar producer properties)

Pulsar 컨슈머 프로퍼티 (Pulsar consumer Properties)

더 알아보기 (Learn more)

• Kafka에서 Pulsar로 옮겨오는 마이그레이션 전략을 더 보고 싶다면 Pulsar Kafka 호환 관련 문서를 참고해요.
• 프로듀서와 컨슈머의 전체 예제는 Pulsar GitHub 리포지토리에서 확인할 수 있어요.
• Pulsar 인증(TLS) 설정에 대해 알아보려면 인증 관련 문서를 살펴보세요.
• 토픽과 구독 개념이 궁금하다면 핵심 개념 문서를 참고해요.