Spring Kafka，识别容器是否为批处理模式答案

【问题标题】：Spring Kafka, identify if container is batch mode or notSpring Kafka，识别容器是否为批处理模式
【发布时间】：2022-02-18 01:46:20
【问题描述】：

commandProcessors-in-0:
  destination: internal-command-processor
  consumer:
    max-attempts: 1
  group: command-processor-group
retrieveCohort-in-0:
  destination: internal-retrieve-cohort
  consumer:
    max-attempts: 1
    batch-mode: true
  group: retrieve-cohort-group

我输入了不同的类似消费者的配置，其中一个可能将批处理模式设置为 true，而另一个则没有。

在我的ListenerContainerCustomizer 中，我想知道他们是否将批处理模式设置为 true。

@Bean
ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer( ){
    return (container, dest, group) -> {
        if (dest.equals("internal-generate-stop-reason")) {
            container.setBatchErrorHandler(new RetryingBatchErrorHandler(new FixedBackOff(5000L, 2L),
                    new DeadLetterPublishingRecoverer(kafkaTemplate(),
                            (rec, ex) -> new TopicPartition("error-dlq", rec.partition()))));
        } else {
            System.out.println(dest+" => "+container.getAssignmentsByClientId());
        }
    };
}

【问题讨论】：

标签： apache-kafka apache-kafka-streams spring-kafka kafka-streams-binder

【解决方案1】：

没有办法知道；除非使用 group/dest 属性。

但是，对于 3.2.x 版本（云 2021.0.x，Apache Kafka 2.8.x 的 Spring），现在有一个 CommonErrorHandler 可以处理批处理和记录侦听器； ErrorHandler 和 BatchErrorHandlers 将在以后的版本中删除。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#eh-summary

【讨论】：

我已经更正了我的答案；您无法检查侦听器类型，因为在调用自定义程序之前未设置它。