Kafka Streams - 覆盖默认 addSink 实现/自定义生产者

Question

这是我在这里的第一篇文章，我不确定之前是否在这里介绍过，但这里是：我有一个 Kafka Streams 应用程序，使用处理器 API，遵循以下拓扑：

1. Consume data from an input topic (processor.addSource())
2. Inserts data into a DB (processor.addProcessor())
3. Produce its process status to an output topic (processor.addSink())

应用程序运行时间很长，但是，出于可追溯性目的，我需要在日志中记录 kstreams 向输出主题生成消息的那一刻，以及它的 RecordMetaData（主题、分区、偏移量）。

示例如下：

KEY="MY_KEY" OUTPUT_TOPIC="MY-OUTPUT-TOPIC" PARTITION="1" OFFSET="1000" STATUS="SUCCESS"

我不确定是否有办法覆盖默认的 kafka 流生产者以添加此日志记录，或者可能创建我自己的生产者以将其插入 addSink 进程。我通过实现自己的 ExceptionHandler (default.producer.exception.handler) 部分实现了它，但它只涵盖了异常。

提前致谢，

吉尔赫姆

Answer 1

如果您将流应用程序配置为使用ProducerInterceptor，那么您应该能够获得所需的信息。具体来说，实施onAcknowledgement() 将提供对您上面列出的所有内容的访问权限。

在流应用程序中配置拦截器：

 Properties props = new Properties();
 // add this configuration in addition to your other streams configs

props.put(StreamsConfig.producerPrefix(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG), Collections.singletonList(MyProducerInterceptor.class));

如果需要，您可以提供多个拦截器，只需添加类名并将列表实现从单例更改为常规List。拦截器的执行遵循列表中类的顺序。

编辑：为了清楚起见，您可以通过KafkaClientSupplier interface 覆盖Kafka Streams 中提供的Producer，但恕我直言，使用拦截器是更清洁的方法。但是要往哪个方向走，就看你自己了。您将KafkaClientSupplier 传递给overloaded Kafka Streams constructor。