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.0</version>
</dependency>

现存的代码无需任何更改即可使用新的依赖。 需要修改的是配置信息，应确保将生产者和消费者使用 Pulsar 服务，而不是Kafka，并使用特定的 Pulsar 主题。

使用 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.0</version>
</dependency>

当使用这个依赖时，应使用 org.apache.kafka.clients.producer.PulsarKafkaProducer 而不是 org.apache.kafka.clients.producer.KafkaProducer 构造生产者，应使用 org.apache.kafka.clients.producer.PulsarKafkaConsumer 构造消费者。

生产者示例

// 主题应为常规 Pulsar 主题
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)

APIs:

属性：

消费者(Consumer)

下述表格列出了consumer 接口

属性：

Customize Pulsar configurations

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

Pulsar client properties

Pulsar producer properties

Pulsar consumer Properties