寻求Hazelcast Jet的Apache Pulsar源连接器及自定义开发指导

Hey there! 确实，目前Hazelcast Jet官方并没有提供针对Apache Pulsar的原生源连接器，我之前在项目里正好需要对接两者，折腾出了一套自定义连接器的方案，分享下我的经验和开发要点，希望能帮到你。

如果只是想快速验证Pulsar和Jet的集成，不用着急做完整的连接器包，可以直接用Pulsar的Java客户端配合Jet的SourceBuilder快速搭建简易源：

• 直接在Jet的源逻辑中初始化Pulsar Consumer，订阅目标主题拉取消息
• 重点注意消息确认、连接重连、分区消费这几个核心点，避免丢消息或者重复消费的问题

1. 基于Jet的Source API构建 #

Hazelcast Jet推荐用SourceBuilder来构建自定义源，比直接实现Source接口更简洁，减少样板代码：

• 核心是调用SourceBuilder.stream("pulsar-source", context -> { ... })，在lambda中完成Pulsar Client和Consumer的初始化
• 通过fillBufferFn定义从Pulsar拉取消息并填充到Jet缓冲区的逻辑
• 用destroyFn处理资源销毁，确保Pulsar Client和Consumer在作业结束时被正确关闭

2. 核心细节处理 #

• 消息确认机制：一定要在Jet处理完消息后再调用Pulsar Consumer的acknowledge方法，避免消息丢失。可以结合Jet的Processors.mapUsingContextAsync，在异步处理完成后执行确认操作
• 分区感知消费：如果Pulsar主题是分区的，要让Jet的分布式节点各自消费对应分区，避免重复消费。通过SourceBuilder.distributed()声明分布式源，然后在初始化时获取Pulsar分区列表，分配给不同的Jet成员
• 故障恢复与重连：Pulsar Consumer自带重连机制，但要在源中捕获连接异常，添加重试逻辑；同时利用Jet的故障转移能力，确保作业重启后能从上次消费的位置继续
• 配置参数化：把Pulsar服务地址、主题名、订阅名、消费者配置（批量拉取大小、超时等）做成可配置项，通过Jet的JobConfig传递，让连接器更通用

3. 代码示例框架 #

这里给一个简化的代码模板，你可以根据实际需求扩展：

public class PulsarSourceProvider {
    public static Source<Message<String>> createPulsarSource(String pulsarServiceUrl, String topic, String subscription) {
        return SourceBuilder.stream("pulsar-stream-source", context -> {
            // 初始化Pulsar Client和Consumer
            PulsarClient client = PulsarClient.builder()
                    .serviceUrl(pulsarServiceUrl)
                    .build();
            Consumer<String> consumer = client.newConsumer(Schema.STRING)
                    .topic(topic)
                    .subscriptionName(subscription)
                    .subscriptionType(SubscriptionType.Shared)
                    .batchReceivePolicy(BatchReceivePolicy.builder().maxNumMessages(100).build())
                    .build();
            
            // 将资源加入Jet的上下文，自动管理关闭
            context.addCloseable(consumer);
            context.addCloseable(client);
            return consumer;
        })
        .fillBufferFn((consumer, buffer) -> {
            // 拉取消息填充到Jet缓冲区
            try {
                List<Message<String>> messages = consumer.batchReceive(100, TimeUnit.MILLISECONDS);
                buffer.addAll(messages);
            } catch (PulsarClientException e) {
                throw new RuntimeException("Failed to fetch messages from Pulsar", e);
            }
        })
        .destroyFn(consumer -> {
            // 确保资源被关闭
            try {
                consumer.close();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
        })
        .distributed() // 开启分布式消费支持
        .build();
    }
}

在Jet作业中使用这个源：

public class PulsarJetJob {
    public static void main(String[] args) {
        JetInstance jet = Jet.bootstrappedInstance();
        Job job = jet.newJob(pipeline -> pipeline
                .readFrom(PulsarSourceProvider.createPulsarSource(
                        "pulsar://localhost:6650", 
                        "my-topic", 
                        "jet-pulsar-sub"
                ))
                .map(Message::getValue)
                .writeTo(Sinks.logger()));
        job.join();
    }
}

4. 开发注意事项 #

• 依赖兼容性：确保Hazelcast Jet和Pulsar Java客户端的版本兼容，比如Jet 5.x搭配Pulsar 2.10+版本比较稳定
• 性能调优：调整Pulsar Consumer的批量拉取大小和Jet的缓冲区容量，平衡吞吐量和延迟；如果是高吞吐量场景，可以开启Pulsar的批量接收模式
• 监控与日志：添加详细日志记录Pulsar连接状态、消息消费数量；利用Jet的内置监控查看源的吞吐量、延迟等指标
• 测试验证：编写单元测试模拟Pulsar消息生产，验证消费逻辑、确认机制；做集成测试时用本地Pulsar集群和Jet集群验证分布式消费场景

• Hazelcast Jet官方文档的「Building Sources」章节，详细讲解了SourceBuilder的使用和自定义源的规范
• Apache Pulsar Java客户端文档，了解Consumer的配置、消息确认、分区处理等核心API
• Hazelcast Jet的官方示例代码，参考其他第三方源的实现思路

内容的提问来源于stack exchange，提问作者vvra