Spring Boot单应用配置多Kafka Stream处理器问题咨询

1. 单应用部署两个Kafka Stream处理器的方案可行性 #

这种方案是Spring Cloud Stream Kafka Streams原生支持的标准实现方式，只要为每个处理器配置独立的application-id和消费组，就可以在单个应用中启动多个互相隔离的Kafka Streams拓扑实例，各自独立维护消费位移、状态存储，从同一个主题消费不存在底层冲突。
但你当前的代码和配置存在多处逻辑错误、配置错误，才会导致第二个处理器运行异常，和方案本身无关。
另外需要注意：你当前需求是两个处理器处理互斥的消息子集，用两个独立拓扑的实现会导致两个处理器各自全量消费输入主题的所有消息，再各自做筛选，会出现重复消费、消息被两个处理器同时命中的问题，并不是最优实现。更合理的方式是定义单个Function，在拓扑内部用branch算子将流按规则拆分成两个分支，分别走窗口聚合、普通处理逻辑，最后合并分支输出到目标主题，性能更好也不会出现消息重复/漏处理。

2. 第二个处理器无法消费消息的诱因与修复方案 #

问题由3个核心错误导致，按影响优先级排序：

• 配置拼写错误：你在配置processorTwo-in-0的消费端参数时，将consumer错写为cosumer（少写字母n），导致该绑定的所有消费端配置（包括DLQ配置、消费参数）完全不生效。在你全局配置了deserialization-exception-handler: sendtodlq的前提下，processorTwo找不到对应的DLQ配置，反序列化异常时无法正常走DLQ投递逻辑，会导致流处理线程反复崩溃重启，根本无法进入正常消费循环。
• 错误的输出端配置：你在两个输出绑定（processorOne-out-0、processorTwo-out-0）上都配置了group参数，消费组是Kafka消费者专属配置，生产者端不需要也不支持该参数，部分Spring Cloud Stream版本会因为输出端存在非法配置导致整个binding生命周期初始化失败，连带输入端无法启动。
• 代码线程安全bug：两个处理器都用方法级的AtomicReference跨算子传递处理结果，Kafka Streams是多线程并行处理模型，不同分区的处理线程会互相覆盖AtomicReference中存储的值，不仅会导致key-value错配，当流处理触发重平衡、异常重试时，还可能出现map算子取到null值的情况，触发Kafka Streams不允许null记录的异常，导致线程挂起。
  另外你的yaml配置中用//写注释是错误的，yaml标准注释符是#，虽然这个配置项是给Kafka Admin用的不影响Streams消费，但还是建议修正避免配置解析异常。

修复步骤 #

1. 把processorTwo-in-0下的cosumer修正为consumer
2. 删除所有*-out-0绑定下的group配置
3. 移除两个处理器中的AtomicReference写法，用无状态算子直接处理数据，比如processorTwo的过滤+转换逻辑可以改成如下写法，不需要外部变量传值：

@Bean
public Function<KStream<String, Input>, KStream<String, Output>> processorTwo() {
  return kStream -> kStream
      .filter((key, value) -> value != null)
      .filter((key, value) -> isValidEvent(value))
      .peek((key, value) -> print(value))
      .flatMap((key, value) -> {
        Optional<Output> outputResult = service.process(value);
        if (outputResult.isPresent()) {
          Output output = outputResult.get();
          print(output);
          String newKey = String.format("%s-%s", output.getNameOne(), output.getNametwo());
          return List.of(KeyValue.pair(newKey, output));
        }
        return List.of();
      });
}

processorOne中对应的AtomicReference逻辑也可以用同样的flatMap方式改造。
4. 确认Serde配置匹配你的POJO类型，如果输入输出是Java对象，不要全局配置StringSerde作为value serde，换成Spring提供的JsonSerde实现，避免不必要的反序列化异常。

3. 单处理器独立配置反序列化异常处理器的方法 #

Spring Cloud Stream Kafka Streams支持按绑定粒度配置反序列化异常处理器，配置优先级高于全局配置，你只需要在对应binding的consumer配置下添加参数即可，示例：

spring:
  cloud:
    stream:
      kafka:
        streams:
          bindings:
            processorOne-in-0:
              consumer:
                dlq-name: processor-one-error-topic
                deserialization-exception-handler: sendtodlq # 单独给processorOne配置
            processorTwo-in-0:
              consumer:
                dlq-name: processor-two-error-topic
                deserialization-exception-handler: logandcontinue # 单独给processorTwo配置，和processorOne做区分

配置时注意不要拼错consumer字段，也不要把该配置写到全局的streams binder配置段下，全局配置会作为所有处理器的默认值，binding下的配置会覆盖默认值。

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