Kafka Streams：当kafka流将数据写入目标主题时如何捕获事件

Question

我正在开发kafka-stream api。基本上Kafka-stream从源主题获取数据，并在应用一些过滤器后将其写回目标kafka主题。

使用的依存关系：

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>2.1.0</version>
    </dependency>

下面是相同的代码。：

{ ...

       // create property

            Properties property =  new Properties();
            property.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
            property.setProperty(StreamsConfig.APPLICATION_ID_CONFIG,"kafka_streams_app");
            property.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
            property.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());

            //create topology
            StreamsBuilder streamsBuilder = new StreamsBuilder();

            //build topology
            KStream<String,String> inputTopic = streamsBuilder.stream("source_topic");

          //filtering data
            KStream<String,String> filteredStream = inputTopic.filter(
                    (k,val)-> filterData(val)>10000
            );
            filteredStream.to("target_topic");
            KafkaStreams streams = new KafkaStreams(
                    streamsBuilder.build(),
                    property
            );
            //start our stream app

            streams.start(); 
    ...
    }

这是我的应用程序架构：

生产者API（源主题中的生产者）=

我想要的是，当流将数据写入目标主题时，我想要捕获事件，无论它是否成功。

有没有办法捕捉到回调？谢谢

Answer 1

您可以在普通生产者API上发送时指定的生产者回调

Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback);

未在Streams API中公开。

不过，可以配置拦截器：

• https://kafka.apache.org/documentation/#producerconfigs_interceptor.classes
• https://javadoc.io/static/org.apache.kafka/kafka-clients/3.0.0/org/apache/kafka/clients/producer/ProducerInterceptor.html