Kafka批处理模式无法设置错误处理器，反序列化报错无限循环求助

问题1：handle方法Consumer参数类型错误 #

• 错误原因：你导入了错误的Consumer类，当前导入的java.util.function.Consumer是Java自带的函数式接口，仅支持1个泛型参数，而SeekToCurrentBatchErrorHandler的handle方法要求的是Kafka客户端自带的消费者类org.apache.kafka.clients.consumer.Consumer，需要2个泛型参数，所以出现参数数量不匹配的报错。
• 解决方法：删除导入语句import java.util.function.Consumer;，替换为import org.apache.kafka.clients.consumer.Consumer;即可。

问题2：Config.this导入歧义与用法错误 #

• 错误原因：
  1. 你当前的配置类类名是StreamConfiguration，代码中写Config.this属于类名引用错误，自然会匹配到其他第三方包下的Config类导致歧义
  2. 你没有在当前配置类中声明ehException成员变量，导致需要依赖外部类的成员变量
• 解决方法：
  1. 在StreamConfiguration类内部添加成员变量声明：private volatile Exception ehException;（加volatile保证多线程可见性）
  2. 将Config.this.ehException = thrownException;修改为StreamConfiguration.this.ehException = thrownException;，不需要导入任何第三方的Config类，直接使用你自己的配置类即可。

附加：反序列化无限循环问题修复 #

你当前使用的SeekToCurrentBatchErrorHandler默认会无限重试失败的消息，反序列化失败的消息会被反复拉取导致死循环，修改错误处理器配置限制重试次数即可，完整配置代码参考：

factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler() {
    {
        // 设置最大重试3次，超过后跳过坏消息
        setMaxAttempts(3);
    }
    @Override
    public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
            MessageListenerContainer container) {
        StreamConfiguration.this.ehException = thrownException;
        super.handle(thrownException, data, consumer, container);
    }
});

另外你的消费者代码中不需要加@RequestBody注解，这是Spring MVC的参数注解，Spring Kafka会自动完成消息反序列化，修改为：

@KafkaListener(id = "#{'${spring.kafka.listener.id}'}", topics = "#{'${spring.kafka.consumer.topic}'}")
public void getTopics(List<Request> model) {
    streamProcessor.runParallel(model.parallelStream());
}

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