跳到主要内容
版本：4.1.1

有什么新特性？

Doubao 中英对照 What’s new?

自 4.0 以来 4.1 的新增内容​

本节介绍从 4.0 版本到 4.1 版本所做的变更。更早版本的变更请参阅变更历史。

@KafkaListener 变更​

@KafkaListener 注解现在支持 ackMode 属性，允许各个监听器覆盖容器工厂的默认确认模式，而无需创建单独的容器工厂 bean。该属性还支持 SpEL 表达式和属性占位符。更多信息请参阅@KafkaListener 注解。

共享消费者确认模式​

ContainerProperties 上的布尔属性 setExplicitShareAcknowledgment(boolean) 已由 ShareAckMode 枚举（EXPLICIT、MANUAL、IMPLICIT）取代。已弃用的方法仍然有效：true 映射到 MANUAL，false 映射到 EXPLICIT。默认行为保持不变——使用默认值的应用程序无需迁移。模式说明请参阅记录确认，升级细节请参阅从 4.0 迁移。

共享消费者异步提交​

ContainerProperties 上新增的 syncShareCommits 属性控制共享容器在确认提交时使用 commitSync()（默认）还是 commitAsync()。当可以接受轻微的确认持久性延迟时，将其设为 false 以获得更高吞吐量。更多信息请参阅同步与异步提交。

共享消费者错误处理​

共享消费者容器现在提供可配置的错误处理：

  • 轮询级：捕获来自 poll() 的 RecordDeserializationException 和 CorruptRecordException，使消费者线程继续运行；无法反序列化的记录被 REJECT，下一次轮询照常进行。
  • 监听器级：一个 ShareConsumerRecordRecoverer 接口在监听器抛出异常时决定是 ACCEPT、RELEASE 还是 REJECT。默认是 ShareConsumerRecordRecoverer.REJECTING；也提供 RELEASING。你可以在工厂或容器上设置自定义 recoverer。
  • 请参阅 Kafka Queues 文档中的共享消费者错误处理。

共享消费者生命周期事件​

ShareKafkaMessageListenerContainer 现在除 ConsumerStartingEvent 和 ConsumerStartedEvent 外，还发布停止和失败生命周期事件：

  • ShareConsumerStoppingEvent（新增）：在底层 ShareConsumer 关闭之前发布。引入一个新的事件类，因为 ShareConsumer 是一个独立于 Consumer 的客户端 API，所以现有的 ConsumerStoppingEvent 无法复用。
  • ConsumerStoppedEvent：在共享消费者关闭后发布，reason 为 NORMAL、ABNORMAL 或 ERROR。
  • ConsumerFailedToStartEvent：当共享消费者启动失败时发布。同一启动尝试中已构造的任何消费者都会被静默关闭。

更多信息请参阅生命周期事件。

ShareAcknowledgment.renew()​

当处理时间超过 broker 的锁持续时间时，ShareAcknowledgment 现在支持 renew() 以延长获取锁（KIP-1222，Kafka 4.2）。更多信息请参阅 Kafka Queues 文档中的 ShareAcknowledgment API。

RetryTopicConfigurationBuilder 默认策略变更​

RetryTopicConfigurationBuilder 中 sameIntervalTopicReuseStrategy 的默认值已从 MULTIPLE_TOPICS 更改为 SINGLE_TOPIC，以与 @RetryableTopic 注解的默认值保持一致。更多信息请参阅主题命名。

Kafka Streams 原生 DLQ 支持​

Spring for Apache Kafka 现在提供支持 Kafka Streams DLQ 的异常处理器（KIP-1034，Kafka 4.2）：

所提供的异常处理器支持多种 Kafka Streams DLQ 启用策略和死信主题解析选项：

RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER 已弃用，取而代之的是 RecoveringDeserializationExceptionHandler.RECOVERER，以与新的处理器属性保持一致。

Kafka Streams 组协议配置​

StreamsBuilderFactoryBean 现在支持配置 groupProtocol 属性。这让你可以显式管理底层 Kafka Streams 消费者是依赖 classic 协议，还是选择更新的 streams 组协议（KIP-1071：Streams 的服务端重平衡）。

更多信息请参阅 Kafka Streams 文档中的配置。

FailedBatchProcessor 中的 BackOffFunction 支持​

FailedBatchProcessor 现在像 FailedRecordProcessor 对单条消息处理那样，为批消息处理提供了 setBackOffFunction 和 setResetStateOnExceptionChange。当使用 DefaultErrorHandler 时，这使得单条处理和批处理之间的异常分类保持一致。

注意，为了保持向后兼容，FailedBatchProcessor 中 resetStateOnExceptionChange 的默认值是 false，而 FailedRecordProcessor 中默认为 true。如果你希望两者行为一致，请在 DefaultErrorHandler 实例上调用 setResetStateOnExceptionChange(true)。

KafkaAdmin 集群 ID 重试间隔​

当 KafkaAdmin.clusterId() 无法从 broker 获取集群 ID 时，现在会记住该失败，在新的 clusterIdRetryInterval 过去之前不再尝试获取；默认为 5 分钟。此前每次调用都会重试，因此当集群 ID 不可用时，启用观察的监听器容器会创建一个 Admin 实例，并在每条记录上阻塞直至操作超时。这改变了默认行为：要选择退出并像以前那样在每次调用时尝试获取，请在 KafkaAdmin 上将 clusterIdRetryInterval 设为 Duration.ZERO。更多信息请参阅Micrometer Observation。