Topic已删除但仍记录Outgoing Enum日志问题求助

问题根源 #

当前代码中使用peek()记录日志，这个操作是在消息进入发送流程之前执行的——不管后续to(targetTopic)是否成功（比如目标Topic已删除导致发送失败），peek()里的日志都会被触发，这就导致了不符合需求的日志记录。

可行方案 #

方案1：使用Producer带回调的send方法 #

替换原流处理链中的peek()+to()，改为在foreach()中直接调用KafkaProducer的send()方法，利用回调确认发送成功后再记录日志：

// 构建或获取与Streams配置一致的KafkaProducer实例
KafkaProducer<K, V> producer = new KafkaProducer<>(streamsConfig);

outputStream.filter((k, v) -> v != null && v.getInput() != null && v.getContent() != null)
           .mapValues(v -> v.getContent())
           .foreach((k, v) -> {
               ProducerRecord<K, V> record = new ProducerRecord<>(targetTopic, k, v);
               // 发送时绑定回调，仅在成功时记录日志
               producer.send(record, (metadata, exception) -> {
                   if (exception == null) {
                       log(enum.getEnumOutgoing(), targetTopic, k);
                   }
                   // 发送失败可按需记录错误日志，此处省略
               });
           });

// 需在KafkaStreams关闭时同步关闭producer，避免资源泄漏
kafkaStreams.setStateListener((newState, oldState) -> {
    if (newState == KafkaStreams.State.NOT_RUNNING) {
        producer.close();
    }
});

方案2：使用ProducerInterceptor拦截器 #

通过实现Kafka的ProducerInterceptor，在消息被Broker确认后触发日志记录，这种方式无需修改流处理链，更贴合Kafka Streams架构：

1. 实现拦截器类：

public class OutgoingLoggingInterceptor<K, V> implements ProducerInterceptor<K, V> {

    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        return record; // 不修改消息，直接传递
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            // 消息发送成功后记录Outgoing日志
            log(enum.getEnumOutgoing(), metadata.topic(), metadata.key());
        }
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

2. 在Streams配置中注册拦截器：

Properties properties = new Properties();
// 其他Streams配置...
properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, 
               OutgoingLoggingInterceptor.class.getName());
StreamsConfig streamsConfig = new StreamsConfig(properties);

注意事项 #

• KafkaProducer是线程安全的，方案1中可安全复用同一个producer实例。
• 方案2的拦截器会被多个Producer线程调用，需确保日志逻辑线程安全。
• 若使用Exactly-Once语义，需考虑重复发送场景下的日志去重需求。

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