Spring集成JavaDSL-Kafka：如何使用Kafka支持将ProducerListener添加到ProducerConfiguration？

Question

我正在开发一个spring集成应用程序，它有一个Kafka出站通道适配器，并使用spring集成java dsl配置流。

• spring集成核心：4.2.4。释放
• spring集成Kafka：1.3.0。释放
• spring集成java dsl：1.1.2。发布

我已经配置了类似于以下代码段的消息处理程序规范。

            KafkaProducerMessageHandlerSpec messageHandlerSpec = Kafka
                .outboundChannelAdapter()
                .addProducer(new ProducerMetadata<String, byte[]>(topicName, String.class, byte[].class,
                        new StringSerializer(), new ByteArraySerializer()), "localhost:9092");

我想添加一个ProducerListener。此功能已添加到此处的Spring集成kafka中。https://github.com/spring-projects/spring-integration-kafka/pull/80

您能为我提供使用Java DSL添加ProducerListener的适当机制吗。

谢谢

Answer 1

嗯，这听起来像是对SI Kafka 1.3的SIJavaDSL的一个很好的改进。现在我们有一个与SI Kafka 1.2. x的兼容性。

与此同时，我们将尝试找出解决方案（请随时提出适当的GH问题），这里有一个解决方法：

KafkaProducerMessageHandlerSpec messageHandlerSpec = Kafka
            .outboundChannelAdapter()
            .addProducer(new ProducerMetadata<String, byte[]>(topicName, String.class, byte[].class,
                    new StringSerializer(), new ByteArraySerializer()), "localhost:9092");

KafkaProducerContext kafkaProducerContext = (KafkaProducerContext) messageHandlerSpec.getComponentsToRegister().iterator().next();
ProducerConfiguration<?, ?> producerConfiguration = kafkaProducerContext.getTopicConfiguration(topicName);
producerConfiguration.setProducerListener(myProducerListener);