Kafka :如何使用、操作、确认(手动)并将结果发送到新主题

Question

我目前正在使用kafka-streams库。

我正在尝试做什么：（使用 kafka 流）我正在尝试从主题消费，操纵消息值，确认该消息并将结果传输到另一个主题（下面的代码）

Properties properties = new Properties();
.
.
properties.put("enable.auto.commit", false);
StreamBuilder builder = new StreamBuilder();
KStream kStream = builder.stream("MyTopic");
KafkaStream kafkaStream = new KafkaStream(builder.build(), properties)
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
kStream.forEach(new ForeachAction<String, String>(){
 @Override
 public void apply(Strings arg, String value){
  //Just doing some simple data manipulation
  String myValue = value + new Date().toString();
  //Sending result to new topic
  producer.send(new ProducerRecord<String,String>("MyTopicWithTimeStamp", myValue)):
  // Problem (1) Here -> How do I acknowledge this from here manually
  // Problem (2) How should I properly handle/close my producer (if at all)
 }
});
kafkaStram.start();

我不知道该怎么做：使用 kafka-streams 库正确确认消息

Answer 1

您可以使用 KStream#to 直接写入 Kafka 输出主题，而不是创建和维护自己的生产者实例，如下所示：

final Properties props = new Properties();
...
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> source = builder.stream("MyTopic");
source.mapValues(record -> record + new Date().toString()).to("MyTopicWithTimeStamp", 
    Produced.with(Serdes.String(), Serdes.String()));
...
final KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Answer 2

您可以按照@amethystic的建议直接通过 kafka stream API 编写，而不是使用手动方法。

source.mapValues(record -> record + new Date().toString()).to("MyTopicWithTimeStamp", 
    Produced.with(Serdes.String(), Serdes.String()));

关于确认，可以在流配置中设置属性，以确保生成者在发布到主题时收到确认。

final Properties props = new Properties();
props .put(StreamsConfig.producerPrefix(ProducerConfig.ACKS_CONFIG), "all");

您可以参考流配置：https://kafka.apache.org/21/documentation/streams/developer-guide/config-streams.html#acks