如何自定义Hazelcast Jet中Sources.jmsQueueBuilder的JMS Session？

我正在使用Hazelcast的Sources.jmsQueueBuilder，功能正常，但需要进一步定制——具体是要自定义jakarta.jms.Session。目前已有connectionFn方法可以创建自定义连接：

connectionFn (connectionFactory to connection)

也提供了consumerFn方法，能从Session创建自定义MessageConsumer，但缺少从Connection到Session的映射器。查看com.hazelcast.jet.impl.connector.StreamJmsP的实现代码，发现它直接硬编码创建Session：

@Override
protected void init(@Nonnull Context context) throws JMSException {
    session = connection.createSession(guarantee != NONE, DUPS_OK_ACKNOWLEDGE);
    consumer = consumerFn.apply(session);
}

完全没有自定义Session的空间。

我的业务场景是：通过JMS（用Sources::jmsQueueBuilder）从RabbitMQ消费消息，再用Hazelcast-Jet做流水线处理，示例代码如下：

StreamSource<Message> jmsQueueSource = Sources.jmsQueueBuilder ....
pipeline.readFrom(jmsQueueSource)
        .withoutTimestamps()
        .map(...) // 自定义映射逻辑
        . /// 其他处理步骤
        .writeTo(Sinks.map("MY_IMAP_NAME")) 

其中MY_IMAP_NAME对应的IMap由DB表通过MapStore/MapLoad提供持久化支持。现在的需求是仅在消息成功保存到DB后，再确认JMS消息。第一步需要把Session的创建逻辑从：

connection.createSession(guarantee != NONE, DUPS_OK_ACKNOWLEDGE);

改为使用CLIENT_ACKNOWLEDGE模式。

更新1：解决方法 #

最终采用的解决步骤如下：

• 先在map-store配置中开启offload: false
• 手动通过SourceBuilder创建自定义流源：

StreamSource<Message> sourceStage = SourceBuilder.stream("streamer", context -> {
     
  RMQConnectionFactory connectionFactory = ...
  RMQConnection connection = (RMQConnection) connectionFactory.createConnection();
  connection.start();
  RMQSession session = (RMQSession) connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
  // 返回包含RMQSession、Connection、Consumer的自定义上下文对象
  
}).<Message>fillBufferFn((context, buf) -> {
  Message message = context.consumer().receiveNoWait();
  if (message != null) {
    buf.add(message);
  } else {
    // 无消息时休眠几秒，避免空轮询
  }
}).destroyFn((context -> {
  context.consumer().close();
  context.session().close();
  context.connection().close();
})).build();

• 基于自定义源构建处理流水线：

pipeline.readFrom(sourceStage)
        .withoutTimestamps()
        .map(m -> map(m)) // 转换为包含JMS消息和JSON内容的记录
        .mapUsingService(ServiceFactories.iMapService("MY_IMAP_NAME"), (iMap, entry) -> {
             // 从记录中提取JSON数据
             // 将JSON存入IMap（触发MapStore写入DB）
             // 从记录中获取JMS消息，手动确认
        })
        .writeTo(Sinks.logger());

更新2：结论 #

原问题的结论正确，现有Hazelcast JMS源API确实无法实现自定义Session的需求，必须通过自定义SourceBuilder的方式解决。

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