Pulsar adaptor for Apache Kafka

Pulsar 为使用 Apache Kafka Java 客户端 API 编写的应用程序提供了一个简单的解决方案。

使用 Pulsar Kafka 兼容封装器

In an existing application, change the regular Kafka client dependency and replace it with the Pulsar Kafka wrapper. Remove the following dependency in pom.xml:

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

引入 Pulsar Kafka 包装器

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

无需更改代码即可使用新的依赖。 需要修改配置信息，并确保将 producer 和 consumer 接入 Pulsar 并使用特定的 Pulsar topic，而不是接入 Kafka。

使用 Pulsar Kafka 兼容封装器和 Kafka 客户端

从 Kafka 迁移到 Pulsar 的过程中，应用程序可能需要同时使用原始 Kafka 客户端和 Pulsar Kafka 封装器。 这种情况下，可以考虑使用透明的 Pulsar Kafka 封装器。

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

当使用这个依赖时，应使用 org.apache.kafka.clients.producer.PulsarKafkaProducer 而不是 org.apache.kafka.clients.producer.KafkaProducer 创建 producer，使用 org.apache.kafka.clients.producer.PulsarKafkaConsumer 创建 consumer。

生产者示例

// 主题应为常规 Pulsar topic
String topic = "persistent://public/default/my-topic";
Properties props = new Properties();
// 指向一个 Pulsar 服务
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();

消费者示例

String topic = "persistent://public/default/my-topic";
Properties props = new Properties();
// 指向一个 Pulsar 服务
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);
    });
    // 提交最近的 offset
    consumer.commitSync();
}

完整示例

点击这里查看完整的生产者和消费者示例

兼容性列表

目前，Pulsar Kafka 封装器支持 Kafka API 提供的多数操作。

生产者（Producer）

API：

属性：

消费者（Consumer）

下表为 consumer API 列表。

属性：

Customize Pulsar configurations

可以通过 Kafka 的属性直接配置 Pulsar 的身份认证。

Pulsar client properties

Pulsar producer properties

Pulsar consumer Properties