跳到主要内容
版本：4.1.1

变更历史

Doubao 中英对照 Change History

自 3.3 以来 4.0 的新增内容​

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

Apache Kafka 4.0 客户端升级​

Spring for Apache Kafka 已升级为使用 Apache Kafka 客户端 4.0.0 版本。此次升级带来了几项重要变更：

  • 所有基于 ZooKeeper 的功能已被移除，因为 Kafka 4.0 已完全过渡到 KRaft 模式
  • ZooKeeper 依赖已从项目中移除
  • 嵌入式 Kafka 测试框架现在仅使用 KRaft 模式
  • EmbeddedKafkaZKBroker 类已被移除，所有功能现在由 EmbeddedKafkaKraftBroker 处理

嵌入式 Kafka 测试框架变更​

测试基础设施已大幅更新：

  • EmbeddedKafkaRule JUnit 4 规则已被移除
  • @EmbeddedKafka 注解已简化，移除了与 ZooKeeper 相关的属性：
  • kraft 属性已被移除，因为 KRaft 模式现在是唯一选项
  • zookeeperPort、zkConnectionTimeout 和 zkSessionTimeout 等 ZooKeeper 专用属性已被移除
  • KafkaClusterTestKit 导入现在使用 KRaft 模式的新包
  • 一些测试已更新，以解决 KRaft 模式下静态端口分配的限制
  • 对测试中的副本因子做了调整，以适应 KRaft 要求

ConsumerRecords 构造器变更​

ConsumerRecords 构造器现在需要一个额外的 Map 参数，框架已对此做了全面处理。直接使用此构造器的应用程序需要更新其代码。

Producer 接口更新​

已实现 Kafka Producer 接口的新方法：

  • registerMetricForSubscription
  • unregisterMetricFromSubscription

移除已弃用功能​

若干已弃用项已被移除：

  • 已弃用的 partitioner 类已从运行时提示中移除
  • 使用 String consumerGroupId 的已弃用 sendOffsetsToTransaction 方法已被移除

Kafka Streams API 变更​

  • KafkaStreamBrancher 已更新，使用新的 split() 和 branch() 方法，而不是已弃用的 branch() 方法
  • DeserializationExceptionHandler 已更新，使用新的 ErrorHandlerContext

与 Apache Kafka 4.0.0 相关的内部 API 更新​

  • BrokerAddress 类现在使用 org.apache.kafka.server.network.BrokerEndPoint，而不是已弃用的 kafka.cluster.BrokerEndPoint
  • GlobalEmbeddedKafkaTestExecutionListener 已更新为仅适用于 KRaft 模式

新消费者重平衡协议​

Spring for Apache Kafka 4.0 支持 Kafka 4.0 的新消费者重平衡协议——KIP-848。详情请参阅新消费者重平衡协议文档。

支持多值头​

JsonKafkaHeaderMapper 和 SimpleKafkaHeaderMapper 支持 Kafka 记录的多值头映射。更多细节请参阅支持多值头映射。

配置附加的 RecordInterceptor​

监听器容器现在支持通过 getRecordInterceptor() 进行拦截器定制。详情请参阅消息监听器容器一节。

批监听器中的逐记录 Observation​

现在使用批监听器时可以为每条记录获取一个 observation。更多信息请参阅批监听器的可观测性。

Kafka Queues（共享消费者）支持​

Spring for Apache Kafka 现在通过共享消费者（share consumer）为 Kafka Queues 提供早期访问支持，它们是 Apache Kafka 4.0.0 的一部分并实现了 KIP-932。这实现了协作式消费，多个消费者可以同时从相同分区消费，与传统的消费者组相比提供了更好的负载分布。更多信息请参阅Kafka Queues（共享消费者）。

Jackson 3 支持​

Spring for Apache Kafka 现在在现有 Jackson 2 支持之外，全面支持 Jackson 3。Jackson 3 会被自动检测，并在可用时优先使用，从而提供增强的性能和现代 JSON 处理能力。

所有 Jackson 2 类现在都有命名一致、类型安全性更好的 Jackson 3 对应类：

  • JsonKafkaHeaderMapper 取代 DefaultKafkaHeaderMapper
  • JacksonJsonSerializer/Deserializer 取代 JsonSerializer/Deserializer
  • JacksonJsonSerde 取代 JsonSerde
  • JacksonJsonMessageConverter 系列取代 JsonMessageConverter 系列
  • JacksonProjectingMessageConverter 取代 ProjectingMessageConverter
  • DefaultJacksonJavaTypeMapper 取代 DefaultJackson2JavaTypeMapper

新的 Jackson 3 类使用 JsonMapper 而非通用的 ObjectMapper 以增强类型安全性，并利用 Jackson 3 改进的模块系统和性能优化。

迁移路径：现有应用程序在 Jackson 2 下继续保持不变。要迁移到 Jackson 3，只需将 Jackson 3 添加到你的类路径中，并更新类引用以使用新的 Jackson 3 对应类。当两个版本同时存在时，框架会自动检测并优先使用 Jackson 3。

向后兼容性：所有 Jackson 2 类已弃用，但仍完全可用。它们将在未来的主版本中移除。

移除 Spring Retry 依赖​

Spring for Apache Kafka 已移除对 Spring Retry 的依赖，转而使用 Spring Framework 7 中引入的核心重试支持。这是一个破坏性变更，影响整个框架中的重试配置和 API。

BackOffValuesGenerator 预先生成所需的 BackOff 值，现在直接使用 Spring Framework 的 BackOff 接口，而非 BackOffPolicy。这些值随后由监听器基础设施管理，不再涉及 Spring Retry。

从配置角度看，Spring Kafka 大量依赖 Spring Retry 的 @Backoff 注解。由于 Spring Framework 中没有等价物，该注解已移至 Spring Kafka 作为 @BackOff，并做了以下改进：

  • 统一命名：使用 @BackOff 而非 @Backoff，以保持一致性
  • 表达式求值：所有字符串属性支持 SpEL 表达式和属性占位符
  • 时长格式支持：字符串属性接受 java.util.Duration 格式（例如 "2s"、"500ms"）
  • 完善文档：改进的 Javadoc，解释更清晰

迁移示例：

// Before
@RetryableTopic(backoff = @Backoff(delay = 2000, maxDelay = 10000, multiplier = 2))

// After
@RetryableTopic(backOff = @BackOff(delay = 2000, maxDelay = 10000, multiplier = 2))

// With new duration format support
@RetryableTopic(backOff = @BackOff(delayString = "2s", maxDelayString = "10s", multiplier = 2))

// With property placeholders
@RetryableTopic(backOff = @BackOff(delayString = "${retry.delay}", multiplierString = "${retry.multiplier}"))

RetryingDeserializer 不再提供 RecoveryCallback，而是提供一个以 RetryException 为输入的等价函数。它包含抛出的异常以及重试次数：

// Before
retryingDeserializer.setRecoveryCallback(context -> {
    return fallbackValue;
});

// After
retryingDeserializer.setRecoveryCallback(retryException -> {
    return fallbackValue;
});

BinaryExceptionClassifier 的使用已被新引入的 ExceptionMatcher 取代，后者提供了更完善的 API。

其他变更包括：

  • DestinationTopicPropertiesFactory 使用 ExceptionMatcher 而非 BinaryExceptionClassifier
  • RetryTopicConfigurationBuilder 中的 uniformRandomBackoff 方法已弃用，转而支持抖动（jitter）支持
  • 错误处理工具已更新，以适配新的异常匹配系统
  • Kafka Streams 重试模板现在使用 Spring Framework 的重试支持

应用程序必须更新其配置以使用新的 Spring Framework 重试 API，但重试行为和功能保持不变。

自 3.2 以来 3.3 的新增内容​

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

DLT 主题命名约定​

DLT 主题的命名约定已标准化为统一使用 "-dlt" 后缀。此变更确保在不同重试方案之间过渡时的兼容性并避免冲突。希望保留 ".DLT" 后缀行为的用户需要通过设置相应的 DLT 名称属性显式启用。

面向消费者组的增强 Seek 操作​

ConsumerSeekCallback 接口新增了一个方法 getGroupId()。该方法允许通过仅定位所需的消费者组来进行更有选择性的 seek 操作。AbstractConsumerSeekAware 现在也可以在多组监听器场景中注册、检索和移除每个主题分区的所有回调而不遗漏。更多详情请参阅新 API（getSeekCallbacksFor(TopicPartition topicPartition)、getTopicsAndCallbacks()）。更多细节请参阅 Seek API 文档。

用 RecordFilterStrategy 的 Kafka 监听器中对空批的可配置处理​

RecordFilterStrategy 现在支持忽略因过滤而产生的空批。这可以通过覆盖默认方法 ignoreEmptyBatch() 来配置，该方法默认为 false，确保即使所有 ConsumerRecords 都被过滤掉也会调用 KafkaListener。更多细节请参阅消息接收过滤文档。

ConcurrentContainerStoppedEvent​

当 ConcurentContainerMessageListenerContainer 的所有子容器都停止时，它现在会发出 ConcurrentContainerStoppedEvent。更多细节请参阅应用事件以及 ConcurrentContainerStoppedEvent Javadoc。

回复中的原始记录键​

使用 ReplyingKafkaTemplate 时，如果请求中的原始记录包含键，那么该相同的键也将成为回复的一部分。更多细节请参阅参考文档的发送消息一节。

在 DeadLetterPublishingRecovererFactory 中定制日志​

使用 DeadLetterPublishingRecovererFactory 时，用户应用程序可以覆盖 maybeLogListenerException 方法以定制日志行为。

在 KafkaAdmin 中定制 Admin 客户端​

扩展 KafkaAdmin 时，用户应用程序可以覆盖 createAdmin 方法以定制 Admin 客户端的创建。

定制 Kafka Streams 的实现​

使用 KafkaStreamsCustomizer 时，现在可以通过覆盖 initKafkaStreams 方法返回 KafkaStreams 对象的自定义实现。

批监听器的 KafkaHeaders.DELIVERY_ATTEMPT​

使用 BatchListener 时，ConsumerRecord 的头字段中可以包含 KafkaHeaders.DELIVERY_ATTMPT 头。如果将 DeliveryAttemptAwareRetryListener 设置为错误处理器作为重试监听器，则每个 ConsumerRecord 都带有投递次数头。更多细节请参阅批监听器的 Kafka 头。

Kafka 指标监听器与 TaskScheduler​

MicrometerProducerListener、MicrometerConsumerListener 和 KafkaStreamsMicrometerListener 现在可以配置 TaskScheduler。更多信息请参阅 KafkaMetricsSupport Javadoc 和Micrometer 支持。

自 3.1 以来 3.2 的新增内容​

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

Kafka 客户端版本​

此版本需要 3.7.0 的 kafka-clients。Kafka 客户端的 3.7.0 版本引入了新的消费者组协议。更多细节及其限制请参阅 KIP-848。新的消费者组协议是早期访问版本，不打算用于生产环境。在此版本中仅建议用于测试目的。因此，Spring for Apache Kafka 对这个新消费者组协议的支持程度仅限于 kafka-client 本身所提供的测试级别支持。默认情况下，Spring for Apache Kafka 使用经典消费者组协议；在测试新的消费者组协议时，需要通过消费者上的 group.protocol 属性显式启用。

测试支持变更​

EmbeddedKafka 中默认禁用 kraft 模式，希望使用 kraft 模式的用户必须显式启用。这是由于在 kraft 模式下使用 EmbeddedKafka 时观察到某些不稳定性，尤其是在测试新消费者组协议时。新的消费者组协议仅在 kraft 模式下受支持，因此在测试新协议时，需要针对真实的 Kafka 集群而非基于 KafkaClusterTestKit（EmbeddedKafka 所基于的）的集群进行测试。此外，在 kraft 模式下使用 EmbeddedKafka 运行多个 KafkaListener 方法时，还观察到一些其他竞态条件。在这些问题解决之前，EmbeddedKafka 上的 kraft 默认将保持为 false。

Kafka Streams 交互式查询支持​

一个用于访问 Kafka Streams 交互式查询中所用可查询存储的新 API KafkaStreamsInteractiveQuerySupport。更多信息请参阅 Kafka Streams 交互式支持。

TransactionIdSuffixStrategy​

引入了一个新的 TransactionIdSuffixStrategy 接口来管理 transactional.id 后缀。默认实现是 DefaultTransactionIdSuffixStrategy，当设置 maxCache 大于零时，可以在特定范围内复用 transactional.id，否则后缀将通过递增计数器动态生成。更多信息请参阅固定 TransactionIdSuffix。

异步 @KafkaListener 返回​

@KafkaListener（和 @KafkaHandler）方法现在可以返回异步返回类型，包括 CompletableFuture<?>、Mono<?> 和 Kotlin suspend 函数。更多信息请参阅异步返回。

基于抛出异常将消息路由到自定义 DLT​

现在可以根据消息处理期间抛出的异常类型，将消息重定向到自定义 DLT。重定向规则通过 RetryableTopic.exceptionBasedDltRouting 或 RetryTopicConfigurationBuilder.dltRoutingRules 设置。自定义 DLT 与其他重试和死信主题一样会自动创建。更多信息请参阅基于抛出异常将消息路由到自定义 DLT。

弃用 ContainerProperties 的 transactionManager 属性​

弃用 ContainerProperties 中的 transactionManager 属性，转而使用 KafkaAwareTransactionManager，与通用的 PlatformTransactionManager 相比这是一个更窄的类型。请参阅 ContainerProperties 和事务同步。

回滚后处理​

提供了一个新的 AfterRollbackProcessor API processBatch。更多信息请参阅回滚后处理器。

更改 @RetryableTopic 的 SameIntervalTopicReuseStrategy 默认值​

将 @RetryableTopic 的 SameIntervalTopicReuseStrategy 属性默认值更改为 SINGLE_TOPIC。请参阅maxInterval 指数延迟的单主题。

非阻塞重试支持类级 @KafkaListener​

非阻塞重试支持类上的 @KafkaListener。请参阅非阻塞重试。

在 RetryTopicConfigurationProvider 中支持处理类上的 @RetryableTopic。​

提供了一个查找 RetryTopicConfiguration 的新公共 API。请参阅查找 RetryTopicConfiguration。

RetryTopicConfigurer 支持处理 MultiMethodKafkaListenerEndpoint。​

RetryTopicConfigurer 支持处理和注册 MultiMethodKafkaListenerEndpoint。MultiMethodKafkaListenerEndpoint 为 defaultMethod 和 methods 属性提供了 getter/setter。修改严格用于 MethodKafkaListenerEndpoint 类型的 EndpointCustomizer。EndpointHandlerMethod 添加了为所提供 bean 构建实例的新构造器。提供了新类 EndpointHandlerMultiMethod，用于为重试端点处理多方法。

基于用户提供的函数 seek 到偏移量的新 API 方法​

ConsumerCallback 提供了一个基于用户定义函数 seek 到偏移量的新 API，该函数以消费者中的当前偏移量为参数。更多信息请参阅 Seek API 文档。

@PartitionOffset 支持 SeekPosition​

为 @PartitionOffset 添加 seekPosition 属性以支持 TopicPartitionOffset.SeekPosition。更多信息请参阅主题分区分配。

TopicPartitionOffset 中接受用于计算要 seek 到的偏移量的函数的新构造器​

TopicPartitionOffset 有一个新构造器，接受一个用户提供的函数来计算要 seek 到的偏移量。使用此构造器时，框架会以当前消费者偏移量位置作为输入参数调用该函数。更多信息请参阅 Seek API 文档。

Spring Boot 应用程序名称作为默认客户端 ID 前缀​

对于定义了应用程序名称的 Spring Boot 应用程序，该名称现在用作某些客户端类型自动生成客户端 ID 的默认前缀。更多信息请参阅默认客户端 ID 前缀。

增强检索 MessageListenerContainer​

ListenerContainerRegistry 提供了两个新 API，可动态查找和过滤 MessageListenerContainer 实例。getListenerContainersMatching(Predicate<String> idMatcher) 用于按 ID 过滤，另一个是 getListenerContainersMatching(BiPredicate<String, MessageListenerContainer> matcher)，用于按 ID 和容器属性过滤。

通过提供更多追踪标签来增强 Observation​

KafkaTemplateObservation 提供更多追踪标签（低基数）。KafkaListenerObservation 提供一个新 API 来查找高基数键名和更多追踪标签（高或低基数）。请参阅 Micrometer Observation。

自 3.0 以来 3.1 的新增内容​

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

Kafka 客户端版本​

此版本需要 3.6.0 的 kafka-clients。

EmbeddedKafkaBroker​

现在提供了一个额外的实现，使用 Kraft 代替 Zookeeper。更多信息请参阅嵌入式 Kafka Broker。

JsonDeserializer​

当发生反序列化异常时，SerializationException 消息不再包含形如 Can’t deserialize data [[123, 34, 98, 97, 122, …​ 的数据；为每个数据字节给出数值数组没有什么用处，对于大数据还会过于冗长。当与 ErrorHandlingDeserializer 一起使用时，发送给错误处理器的 DeserializationException 包含 data 属性，其中存有无法反序列化的原始数据。当不与 ErrorHandlingDeserializer 一起使用时，KafkaConsumer 会持续为同一条记录抛出异常，显示主题/分区/偏移量以及 Jackson 抛出的原因。

ContainerPostProcessor​

通过在 @KafkaListener 注解上指定一个 ContainerPostProcessor 的 bean 名称，可以对监听器容器应用后处理。这发生在容器创建之后，以及容器工厂上配置的任何 ContainerCustomizer 之后。更多信息请参阅容器工厂。

ErrorHandlingDeserializer​

现在你可以向该反序列化器添加一个 Validator；如果委托的 Deserializer 成功反序列化了对象，但该对象校验失败，则抛出一个类似于反序列化异常的异常。这样可以将原始数据传递给错误处理器。更多信息请参阅使用 ErrorHandlingDeserializer。

可重试主题​

当 @RetryableTopic(backOff = @BackOff(delay = 5000), attempts = "2", fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC) 时，将后缀 -retry-5000 更改为 -retry。如果你想保留后缀 -retry-5000，请使用 @RetryableTopic(backOff = @BackOff(delay = 5000), attempts = "2")。更多信息请参阅主题命名。

监听器容器变更​

当手动分配分区且消费者 group.id 为 null 时，AckMode 现在会被自动强制设为 MANUAL。更多信息请参阅手动分配所有分区。

自 2.9 以来 3.0 的新增内容​

Kafka 客户端版本​

此版本需要 3.3.1 的 kafka-clients。

恰好一次语义​

不再支持 EOSMode.V1（即 ALPHA）。

important

使用事务时，最低 broker 版本为 2.5。

更多信息请参阅恰好一次语义和 KIP-447。

Observation​

现在支持使用 Micrometer 为计时器和链路追踪启用观察。更多信息请参阅 Observation。

原生镜像​

提供了创建原生镜像的支持。更多信息请参阅原生镜像。

全局单一嵌入式 Kafka​

嵌入式 Kafka（EmbeddedKafkaBroker）现在可以作为整个测试计划的单一全局实例启动。更多信息请参阅为多个测试类使用相同的 Broker。

可重试主题变更​

此功能不再被视为实验性（就其 API 而言），该功能本身自 2.7 起已受支持，但出现破坏性 API 变更的可能性高于正常水平。

本发行版更改了非阻塞重试基础设施 bean 的引导方式，以避免某些应用在应用初始化时出现的时序问题。

现在你可以为重试容器设置不同的 concurrency；默认情况下，并发度与主容器相同。

@RetryableTopic 现在可以在自定义注解上用作元注解，包括对 @AliasFor 属性的支持。

更多信息请参阅配置。

重试主题的默认副本因子现在为 -1（使用 broker 默认值）。如果你的 broker 早于 2.4 版本，现在需要显式设置该属性。

现在你可以在同一个应用上下文中为同一主题配置多个 @RetryableTopic 监听器。此前这是不可能的。更多信息请参阅多个监听器、相同主题。

RetryTopicConfigurationSupport 存在破坏性 API 变更；具体来说，如果你覆盖了 destinationTopicResolver、kafkaConsumerBackoffManager 和/或 retryTopicConfigurer 的 bean 定义方法；这些方法现在需要一个 ObjectProvider<RetryTopicComponentFactory> 参数。

监听器容器变更​

与消费者认证和授权失败相关的事件现在由容器发布。更多信息请参阅应用事件。

现在你可以定制消费者线程所使用的线程名称。更多信息请参阅容器线程命名。

新增了容器属性 restartAfterAuthException。更多信息请参阅监听器容器属性。

KafkaTemplate 变更​

此类返回的 future 现在是 CompletableFuture 而非 ListenableFuture。更多信息请参阅使用 KafkaTemplate。

ReplyingKafkaTemplate 变更​

此类返回的 future 现在是 CompletableFuture 而非 ListenableFuture。更多信息请参阅使用 ReplyingKafkaTemplate和使用 Message<?> 请求/回复。

@KafkaListener 变更​

现在你可以使用一个自定义关联头，它将在任何回复消息中回显。更多信息请参阅使用 ReplyingKafkaTemplate 末尾的备注。

现在你可以在整个批处理完成之前手动提交批中的部分内容。更多信息请参阅提交偏移量。

KafkaHeaders 变更​

KafkaHeaders 中在 2.9.x 已弃用的四个常量现已被移除。

  • 使用 KEY 代替 MESSAGE_KEY。
  • 使用 PARTITION 代替 PARTITION_ID

类似地，RECEIVED_MESSAGE_KEY 由 RECEIVED_KEY 取代，RECEIVED_PARTITION_ID 由 RECEIVED_PARTITION 取代。

测试变更​

3.0.7 版本引入了 MockConsumerFactory 和 MockProducerFactory。更多信息请参阅Mock 消费者与生产者。

从 3.0.10 版本开始，嵌入式 Kafka broker 默认将 Spring Boot 属性 spring.kafka.bootstrap-servers 设置为嵌入式 broker 的地址。

自 2.8 以来 2.9 的新增内容​

Kafka 客户端版本​

此版本需要 3.2.0 的 kafka-clients。

错误处理器变更​

DefaultErrorHandler 现在可以配置为让容器暂停一次轮询，并使用上一次轮询的剩余结果，而不是 seek 到剩余记录的偏移量。更多信息请参阅DefaultErrorHandler。

DefaultErrorHandler 现在有一个 BackOffHandler 属性。更多信息请参阅退避处理器。

监听器容器变更​

interceptBeforeTx 现在适用于所有事务管理器（此前仅在使用 KafkaAwareTransactionManager 时应用）。更多信息请参阅 interceptBeforeTx。

新增了一个容器属性 pauseImmediate，它允许容器在当前记录处理完成后立即暂停消费者，而不是在上一次轮询的所有记录都处理完后才暂停。更多信息请参阅 pauseImmediate。

与消费者认证和授权相关的事件。

头映射器变更​

现在你可以配置哪些入站头应被映射。2.8.8 及更高版本中也可用。更多信息请参阅消息头。

KafkaTemplate 变更​

在 3.0 中，此类返回的 future 将是 CompletableFuture 而非 ListenableFuture。使用本发行版时，更多信息请参阅使用 KafkaTemplate 以协助过渡。

ReplyingKafkaTemplate 变更​

该模板现在提供了一个方法，用于等待回复容器上的分配，以避免在回复容器初始化之前发送请求时出现竞态。2.8.8 及更高版本中也可用。更多信息请参阅使用 ReplyingKafkaTemplate。

在 3.0 中，此类返回的 future 将是 CompletableFuture 而非 ListenableFuture。使用本发行版时，更多信息请参阅使用 ReplyingKafkaTemplate和使用 Message<?> 请求/回复 以协助过渡。

自 2.7 以来 2.8 的新增内容​

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

Kafka 客户端版本​

此版本需要 3.0.0 的 kafka-clients。

包变更​

与类型映射相关的类和接口已从 …​support.converter 移至 …​support.mapping。

  • AbstractJavaTypeMapper
  • ClassMapper
  • DefaultJackson2JavaTypeMapper
  • Jackson2JavaTypeMapper

乱序手动提交​

现在可以配置监听器容器以接受乱序（通常是异步的）手动偏移量提交。容器将延迟提交，直到缺失的偏移量被确认。更多信息请参阅手动提交偏移量。

@KafkaListener 变更​

现在可以在监听器方法本身上指定该方法是否为批监听器。这使得同一个容器工厂既可用于记录监听器，也可用于批监听器。

更多信息请参阅批监听器。

批监听器现在可以处理转换异常。

RecordFilterStrategy 在与批监听器一起使用时，现在可以在一次调用中过滤整个批。更多信息请参阅批监听器 末尾的备注。

@KafkaListener 注解现在有了 filter 属性，可仅针对此监听器覆盖容器工厂的 RecordFilterStrategy。

@KafkaListener 注解现在有了 info 属性；它用于填充新的监听器容器属性 listenerInfo。该属性随后用于在每条记录中填充 KafkaHeaders.LISTENER_INFO 头，可在 RecordInterceptor、RecordFilterStrategy 或监听器本身中使用。更多信息请参阅监听器信息头和AbstractMessageListenerContainer 属性。

KafkaTemplate 变更​

现在你可以根据主题、分区和偏移量接收单条记录。更多信息请参阅使用 KafkaTemplate 接收。

新增 CommonErrorHandler​

用于记录和批监听器的遗留 GenericErrorHandler 及其子接口层次结构，已被新的单一接口 CommonErrorHandler 取代，其实现对应于 GenericErrorHandler 的大多数遗留实现。更多信息请参阅容器错误处理器和将自定义遗留错误处理器实现迁移到 CommonErrorHandler。

监听器容器变更​

interceptBeforeTx 容器属性现在默认为 true。

authorizationExceptionRetryInterval 属性已重命名为 authExceptionRetryInterval，并且现在除了此前的 AuthorizationException 之外，还适用于 AuthenticationException。这两种异常都被视为致命异常，除非设置此属性，否则容器默认会停止。

序列化器/反序列化器变更​

现在提供了 DelegatingByTopicSerializer 和 DelegatingByTopicDeserializer。更多信息请参阅委托序列化器与反序列化器。

DeadLetterPublishingRecover 变更​

属性 stripPreviousExceptionHeaders 现在默认为 true。

现在有若干技术可用于定制添加到输出记录的头。

更多信息请参阅管理死信记录头。

可重试主题变更​

现在你可以对可重试和不可重试主题使用同一个工厂。更多信息请参阅指定 ListenerContainerFactory。

现在有一个可管理的全局致命异常列表，会使失败的记录直接进入 DLT。如何管理它请参阅异常分类器。

现在你可以组合使用阻塞和非阻塞重试。更多信息请参阅组合阻塞与非阻塞重试。

使用可重试主题功能时抛出的 KafkaBackOffException 现在以 DEBUG 级别记录日志。如果你需要将日志级别改回 WARN 或设为任何其他级别，请参阅更改 KafkaBackOffException 日志级别。

2.6 与 2.7 之间的变更​

Kafka 客户端版本​

此版本需要 2.7.0 的 kafka-clients。自 2.7.1 起，它也兼容 2.8.0 客户端；请参阅覆盖 Spring Boot 依赖。

使用主题进行非阻塞延迟重试​

本发行版新增了这一重要功能。当严格的顺序不重要时，投递失败的消息可以被发送到另一个主题，稍后再消费。可以配置一系列这样的重试主题，延迟递增。更多信息请参阅非阻塞重试。

监听器容器变更​

onlyLogRecordMetadata 容器属性现在默认为 true。

现在提供了一个新的容器属性 stopImmediate。

更多信息请参阅监听器容器属性。

在投递尝试之间使用 BackOff 的错误处理器（例如 SeekToCurrentErrorHandler 和 DefaultAfterRollbackProcessor），现在会在容器停止后不久退出退避间隔，而不是延迟停止。

扩展了 FailedRecordProcessor 的错误处理器和回滚后处理器，现在可以配置一个或多个 RetryListener，以接收关于重试和恢复进度的信息。

RecordInterceptor 现在有一些在监听器返回（正常返回或抛出异常）后调用的额外方法。它还有一个子接口 ConsumerAwareRecordInterceptor。此外，现在有了用于批监听器的 BatchInterceptor。更多信息请参阅消息监听器容器。

@KafkaListener 变更​

现在你可以校验 @KafkaHandler 方法（类级监听器）的负载参数。更多信息请参阅@KafkaListener @Payload 校验。

现在你可以在 MessagingMessageConverter 和 BatchMessagingMessageConverter 上设置 rawRecordHeader 属性，它会使原始 ConsumerRecord 被添加到转换后的 Message<?> 中。例如，如果你希望在监听器错误处理器中使用 DeadLetterPublishingRecoverer，这会很有用。更多信息请参阅监听器错误处理器。

现在你可以在应用初始化期间修改 @KafkaListener 注解。更多信息请参阅@KafkaListener 属性修改。

DeadLetterPublishingRecover 变更​

现在，如果键和值都反序列化失败，原始值会被发布到 DLT。此前，值会被填充，但键的 DeserializationException 仍保留在头中。如果你子类化了该 recoverer 并覆盖了 createProducerRecord 方法，则存在破坏性 API 变更。

此外，recoverer 会在发布到某个分区之前，验证目的地解析器所选的分区是否真实存在。

更多信息请参阅发布死信记录。

ChainedKafkaTransactionManager 已弃用​

更多信息请参阅事务。

ReplyingKafkaTemplate 变更​

现在有了一种机制，可以检查回复并在存在某种条件时以异常完成 future。

新增了对发送和接收 spring-messaging Message<?> 的支持。

更多信息请参阅使用 ReplyingKafkaTemplate。

Kafka Streams 变更​

默认情况下，StreamsBuilderFactoryBean 现在被配置为不清理本地状态。更多信息请参阅配置。

KafkaAdmin 变更​

新增了 createOrModifyTopics 和 describeTopics 方法。新增了 KafkaAdmin.NewTopics，以便在单个 bean 中配置多个主题。更多信息请参阅配置主题。

MessageConverter 变更​

现在可以向 MessagingMessageConverter 添加一个 spring-messaging SmartMessageConverter，从而基于 contentType 头进行内容协商。更多信息请参阅Spring Messaging 消息转换。

@KafkaListener 的排序​

更多信息请参阅按顺序启动 @KafkaListener。

ExponentialBackOffWithMaxRetries​

提供了一个新的 BackOff 实现，使配置最大重试次数更加方便。更多信息请参阅ExponentialBackOffWithMaxRetries 实现。

条件委托错误处理器​

这些新的错误处理器可以配置为根据异常类型委托给不同的错误处理器。更多信息请参阅委托错误处理器。

2.5 与 2.6 之间的变更​

Kafka 客户端版本​

此版本需要 2.6.0 的 kafka-clients。

监听器容器变更​

默认的 EOSMode 现在为 BETA。更多信息请参阅恰好一次语义。

各种错误处理器（扩展了 FailedRecordProcessor）和 DefaultAfterRollbackProcessor 现在在恢复失败时会重置 BackOff。此外，你现在可以根据失败的记录和/或异常选择要使用的 BackOff。

现在你可以在容器属性中配置 adviceChain。更多信息请参阅监听器容器属性。

当容器被配置为发布 ListenerContainerIdleEvent 时，现在在发布空闲事件之后收到记录时，它会发布一个 ListenerContainerNoLongerIdleEvent。更多信息请参阅应用事件和检测空闲和无响应消费者。

@KafkaListener 变更​

使用手动分区分配时，现在你可以指定一个通配符，用于决定哪些分区应重置为初始偏移量。此外，如果监听器实现了 ConsumerSeekAware，则在手动分配之后会调用 onPartitionsAssigned()。（也在 2.5.5 版本中加入。）更多信息请参阅主题分区分配。

AbstractConsumerSeekAware 中新增了一些便捷方法，使 seek 操作更简单。更多信息请参阅Seek 到特定偏移量。

ErrorHandler 变更​

FailedRecordProcessor 的子类（例如 SeekToCurrentErrorHandler、DefaultAfterRollbackProcessor、RecoveringBatchErrorHandler）现在可以配置为：当异常类型与该记录此前发生的异常不同时，重置重试状态。

生产者工厂变更​

现在你可以为生产者设置最大存活时间，超过后它们将被关闭并重新创建。更多信息请参阅事务。

现在你可以在 DefaultKafkaProducerFactory 创建之后更新其配置映射。例如，当你需要在凭据变更后更新 SSL 密钥/信任库位置时，这可能很有用。更多信息请参阅使用 DefaultKafkaProducerFactory。

2.4 与 2.5 之间的变更​

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

消费者/生产者工厂变更​

默认的消费者和生产者工厂现在可以在每次创建或关闭消费者/生产者时调用一个回调。提供了用于原生 Micrometer 指标的实现。更多信息请参阅工厂监听器。

现在你可以在运行时更改引导服务器属性，从而实现故障转移到另一个 Kafka 集群。更多信息请参阅连接到 Kafka。

StreamsBuilderFactoryBean 变更​

该工厂 bean 现在可以在每次创建或销毁 KafkaStreams 时调用一个回调。提供了用于原生 Micrometer 指标的实现。更多信息请参阅KafkaStreams Micrometer 支持。

Kafka 客户端版本​

此版本需要 2.5.0 的 kafka-clients。

类/包变更​

SeekUtils 已从 o.s.k.support 包移至 o.s.k.listener。

投递次数头​

现在可以选择添加一个头，在使用某些错误处理器和回滚后处理器时跟踪投递次数。更多信息请参阅投递次数头。

@KafkaListener 变更​

当 @KafkaListener 返回类型为 Message<?> 时，现在会在需要时自动填充默认回复头。更多信息请参阅回复类型 Message<?>。

当入站记录的键为 null 时，KafkaHeaders.RECEIVED_MESSAGE_KEY 不再以 null 值填充；该头会被完全省略。

@KafkaListener 方法现在可以指定一个 ConsumerRecordMetadata 参数，而不必使用离散的头来表示主题、分区等元数据。更多信息请参阅消费者记录元数据。

监听器容器变更​

assignmentCommitOption 容器属性现在默认为 LATEST_ONLY_NO_TX。更多信息请参阅监听器容器属性。

使用事务时，subBatchPerPartition 容器属性现在默认为 true。更多信息请参阅事务。

现在提供了一个新的 RecoveringBatchErrorHandler。

现在支持静态组成员关系。更多信息请参阅消息监听器容器。

当配置了增量/协作式重平衡时，如果以非致命的 RebalanceInProgressException 提交偏移量失败，容器将尝试在重平衡完成后，为仍分配给该实例的分区重新提交偏移量。

默认错误处理器现在是：对记录监听器使用 SeekToCurrentErrorHandler，对批监听器使用 RecoveringBatchErrorHandler。更多信息请参阅容器错误处理器。

现在你可以控制标准错误处理器有意抛出的异常的日志级别。更多信息请参阅容器错误处理器。

新增了 getAssignmentsByClientId() 方法，更容易确定并发容器中的哪些消费者被分配到了哪些分区。更多信息请参阅监听器容器属性。

现在你可以在 error、debug 等日志中抑制记录整个 ConsumerRecord。请参阅监听器容器属性中的 onlyLogRecordMetadata。

KafkaTemplate 变更​

KafkaTemplate 现在可以维护 micrometer 计时器。更多信息请参阅监控。

KafkaTemplate 现在可以用 ProducerConfig 属性配置，以覆盖生产者工厂中的相应属性。更多信息请参阅使用 KafkaTemplate。

现在提供了一个 RoutingKafkaTemplate。更多信息请参阅使用 RoutingKafkaTemplate。

现在你可以使用 KafkaSendCallback 代替 ListenerFutureCallback 来获得一个更窄的异常，从而更容易提取失败的 ProducerRecord。更多信息请参阅使用 KafkaTemplate。

Kafka 字符串序列化器/反序列化器​

现在提供了新的 ToStringSerializer/StringDeserializer 以及一个相关的 SerDe。更多信息请参阅字符串序列化。

JsonDeserializer​

JsonDeserializer 现在有更大的灵活性来确定反序列化类型。更多信息请参阅使用方法确定类型。

委托序列化器/反序列化器​

当出站记录没有头时，DelegatingSerializer 现在可以处理"标准"类型。更多信息请参阅委托序列化器与反序列化器。

测试变更​

KafkaTestUtils.consumerProps() 辅助记录现在默认将 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG 设置为 earliest。更多信息请参阅 JUnit。

2.3 与 2.4 之间的变更​

Kafka 客户端版本​

此版本需要 2.4.0 或更高版本的 kafka-clients，并支持新的增量重平衡功能。

ConsumerAwareRebalanceListener​

与 ConsumerRebalanceListener 一样，此接口现在新增了一个方法 onPartitionsLost。更多信息请参阅 Apache Kafka 文档。

与 ConsumerRebalanceListener 不同，默认实现不调用 onPartitionsRevoked。相反，监听器容器会在调用 onPartitionsLost 之后再调用该方法；因此，在实现 ConsumerAwareRebalanceListener 时你不应重复这样做。

更多信息请参阅重平衡监听器末尾的重要备注。

GenericErrorHandler​

isAckAfterHandle() 的默认实现现在默认为 true。

KafkaTemplate​

KafkaTemplate 现在除事务性发布外还支持非事务性发布。更多信息请参阅KafkaTemplate 事务性与非事务性发布。

AggregatingReplyingKafkaTemplate​

releaseStrategy 现在是一个 BiConsumer。它现在会在超时之后（以及记录到达时）被调用；在超时后调用的情况下，第二个参数为 true。

更多信息请参阅聚合多个回复。

监听器容器​

ContainerProperties 提供了一个 authorizationExceptionRetryInterval 选项，让监听器容器在 KafkaConsumer 抛出任何 AuthorizationException 之后重试。更多信息请参阅其 Javadoc 和使用 KafkaMessageListenerContainer。

@KafkaListener​

@KafkaListener 注解有一个新属性 splitIterables；默认 true。当回复监听器返回一个 Iterable 时，该属性控制返回结果是作为单条记录发送，还是为每个元素发送一条记录。更多信息请参阅使用 @SendTo 转发监听器结果。

现在可以为批监听器配置一个 BatchToRecordAdapter；例如，这允许批在一个事务中处理，同时监听器一次获取一条记录。使用默认实现时，可以使用 ConsumerRecordRecoverer 处理批内的错误，而不必停止整个批的处理——这在使用事务时可能很有用。更多信息请参阅使用批监听器的事务。

Kafka Streams​

StreamsBuilderFactoryBean 接受一个新属性 KafkaStreamsInfrastructureCustomizer。这允许在流创建之前配置构建器和/或拓扑。更多信息请参阅Spring 管理。

2.2 与 2.3 之间的变更​

本节介绍从 2.2 版本到 2.3 版本所做的变更。

技巧、窍门与示例​

新增了一个章节技巧、窍门与示例。请通过提交 GitHub issue 和/或 pull request 为该章节补充更多条目。

Kafka 客户端版本​

此版本需要 2.3.0 或更高版本的 kafka-clients。

类/包变更​

TopicPartitionInitialOffset 已弃用，取而代之的是 TopicPartitionOffset。

配置变更​

从 2.3.4 版本开始，missingTopicsFatal 容器属性默认为 false。当此属性为 true 时，如果 broker 宕机，应用将启动失败；许多用户受到了这一变更的影响；鉴于 Kafka 是一个高可用平台，我们此前没有预料到在没有活动 broker 的情况下启动应用会是一个常见用例。

生产者与消费者工厂变更​

现在可以配置 DefaultKafkaProducerFactory 为每个线程创建一个生产者。你也可以在构造器中提供 Supplier<Serializer> 实例，作为对配置类（需要无参构造器）或使用 Serializer 实例构造（随后所有生产者共享）的替代。更多信息请参阅使用 DefaultKafkaProducerFactory。

DefaultKafkaConsumerFactory 中同样可以使用 Supplier<Deserializer> 实例。更多信息请参阅使用 KafkaMessageListenerContainer。

监听器容器变更​

此前，当使用监听器适配器（例如 @KafkaListener）调用监听器时，错误处理器会收到 ListenerExecutionFailedException（实际监听器异常作为 cause）。原生 GenericMessageListener 抛出的异常则原样传递给错误处理器。现在，参数始终是 ListenerExecutionFailedException（实际监听器异常作为 cause），从而可以访问容器的 group.id 属性。

由于监听器容器有自己提交偏移量的机制，它倾向于将 Kafka 的 ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG 设为 false。现在除非在消费者工厂中显式设置，或容器的消费者属性覆盖，它会自动将其设为 false。

ackOnError 属性现在默认为 false。

现在可以在监听器方法中获取消费者的 group.id 属性。更多信息请参阅获取消费者 group.id。

容器新增了一个 recordInterceptor 属性，允许在调用监听器之前检查或修改记录。如果你需要调用多个拦截器，还提供了 CompositeRecordInterceptor。更多信息请参阅消息监听器容器。

ConsumerSeekAware 新增了一些方法，允许你相对于起始、末尾或当前位置执行 seek，以及 seek 到大于或等于某个时间戳的第一个偏移量。更多信息请参阅Seek 到特定偏移量。

现在提供了一个便捷类 AbstractConsumerSeekAware 以简化 seek 操作。更多信息请参阅Seek 到特定偏移量。

ContainerProperties 提供了一个 idleBetweenPolls 选项，让监听器容器中的主循环在 KafkaConsumer.poll() 调用之间休眠。更多信息请参阅其 Javadoc 和使用 KafkaMessageListenerContainer。

使用 AckMode.MANUAL（或 MANUAL_IMMEDIATE）时，现在你可以通过在 Acknowledgment 上调用 nack 来导致重新投递。更多信息请参阅提交偏移量。

现在可以使用 Micrometer Timer 监控监听器性能。更多信息请参阅监控。

容器现在会发布与启动相关的额外消费者生命周期事件。更多信息请参阅应用事件。

事务性批监听器现在可以支持僵尸隔离（zombie fencing）。更多信息请参阅事务。

监听器容器工厂现在可以配置一个 ContainerCustomizer，以便在每个容器创建并配置之后进一步配置它。更多信息请参阅容器工厂。

ErrorHandler 变更​

SeekToCurrentErrorHandler 现在将某些异常视为致命，并对其禁用重试，在首次失败时即调用 recoverer。

SeekToCurrentErrorHandler 和 SeekToCurrentBatchErrorHandler 现在可以配置为在投递尝试之间应用 BackOff（线程休眠）。

从 2.3.2 版本开始，当错误处理器在恢复一条失败记录之后返回时，将提交已恢复记录的偏移量。

当 DeadLetterPublishingRecoverer 与 ErrorHandlingDeserializer 一起使用时，现在会将发送到死信主题的消息负载设置为无法反序列化的原始值。此前该值为 null，用户代码需要从消息头中提取 DeserializationException。更多信息请参阅发布死信记录。

TopicBuilder​

新增了一个类 TopicBuilder，用于更方便地创建 NewTopic @Bean 以自动供应主题。更多信息请参阅配置主题。

Kafka Streams 变更​

现在你可以对 @EnableKafkaStreams 创建的 StreamsBuilderFactoryBean 进行额外配置。更多信息请参阅Streams 配置。

现在提供了一个 RecoveringDeserializationExceptionHandler，允许恢复带有反序列化错误的记录。它可以与 DeadLetterPublishingRecoverer 一起使用，将这些记录发送到死信主题。更多信息请参阅从反序列化异常中恢复。

提供了 HeaderEnricher 转换器，使用 SpEL 生成头值。更多信息请参阅Header Enricher。

提供了 MessagingTransformer。它允许 Kafka 流拓扑与 spring-messaging 组件（例如 Spring Integration 流）交互。更多信息请参阅 MessagingProcessor 和从 KStream 调用 Spring Integration 流。

JSON 组件变更​

现在所有 JSON 感知组件默认都配置了由 JacksonUtils.enhancedObjectMapper() 产生的 Jackson ObjectMapper。JsonDeserializer 现在提供基于 TypeReference 的构造器，以便更好地处理目标泛型容器类型。此外，还引入了 JacksonMimeTypeModule，用于将 org.springframework.util.MimeType 序列化为纯字符串。更多信息请参阅其 Javadoc 和序列化、反序列化与消息转换。

还提供了 ByteArrayJsonMessageConverter，以及所有 Json 转换器的一个新超类 JsonMessageConverter。此外，现在有了 StringOrBytesSerializer；它可以序列化 ProducerRecord 中的 byte[]、Bytes 和 String 值。更多信息请参阅Spring Messaging 消息转换。

JsonSerializer、JsonDeserializer 和 JsonSerde 现在有了流式 API，使编程式配置更简单。更多信息请参阅 Javadoc、序列化、反序列化与消息转换和Streams JSON 序列化与反序列化。

ReplyingKafkaTemplate​

当回复超时时，future 会以 KafkaReplyTimeoutException 而非 KafkaException 异常完成。

此外，现在提供了一个重载的 sendAndReceive 方法，允许按消息指定回复超时。

AggregatingReplyingKafkaTemplate​

通过聚合来自多个接收者的回复来扩展 ReplyingKafkaTemplate。更多信息请参阅聚合多个回复。

事务变更​

现在你可以在 KafkaTemplate 和 KafkaTransactionManager 上覆盖生产者工厂的 transactionIdPrefix。更多信息请参阅transactionIdPrefix。

新的委托序列化器/反序列化器​

框架现在提供了一个委托 Serializer 和 Deserializer，利用一个头来支持生产和消费具有多种键/值类型的记录。更多信息请参阅委托序列化器与反序列化器。

新的重试反序列化器​

框架现在提供了一个委托 RetryingDeserializer，用于在可能发生网络问题等瞬时错误时重试序列化。更多信息请参阅重试反序列化器。

2.1 与 2.2 之间的变更​

Kafka 客户端版本​

此版本需要 2.0.0 或更高版本的 kafka-clients。

类与包变更​

ContainerProperties 类已从 org.springframework.kafka.listener.config 移至 org.springframework.kafka.listener。

AckMode 枚举已从 AbstractMessageListenerContainer 移至 ContainerProperties。

setBatchErrorHandler() 和 setErrorHandler() 方法已从 ContainerProperties 移至 AbstractMessageListenerContainer 和 AbstractKafkaListenerContainerFactory。

回滚后处理​

提供了一个新的 AfterRollbackProcessor 策略。更多信息请参阅回滚后处理器。

ConcurrentKafkaListenerContainerFactory 变更​

现在你可以使用 ConcurrentKafkaListenerContainerFactory 创建和配置任何 ConcurrentMessageListenerContainer，而不仅仅是用于 @KafkaListener 注解的容器。更多信息请参阅容器工厂。

监听器容器变更​

新增了一个容器属性（missingTopicsFatal）。更多信息请参阅使用 KafkaMessageListenerContainer。

当消费者停止时，现在会发出 ConsumerStoppedEvent。更多信息请参阅线程安全。

批监听器现在可以选择接收完整的 ConsumerRecords<?, ?> 对象，而不是 List<ConsumerRecord<?, ?>>。更多信息请参阅批监听器。

DefaultAfterRollbackProcessor 和 SeekToCurrentErrorHandler 现在可以恢复（跳过）持续失败的记录，默认在 10 次失败后这样做。它们可以配置为将失败的记录发布到死信主题。

从 2.2.4 版本开始，在选择死信主题名称时可以使用消费者的组 ID。

新增了 ConsumerStoppingEvent。更多信息请参阅应用事件。

当容器配置为 AckMode.MANUAL_IMMEDIATE 时（自 2.2.4 起），SeekToCurrentErrorHandler 现在可以配置为提交已恢复记录的偏移量。

@KafkaListener 变更​

现在你可以通过在注解上设置属性来覆盖监听器容器工厂的 concurrency 和 autoStartup 属性。现在你可以添加配置，决定哪些头（如果有）被复制到回复消息。更多信息请参阅@KafkaListener 注解。

现在你可以在自己的注解上把 @KafkaListener 用作元注解。更多信息请参阅@KafkaListener 作为元注解。

现在更容易为 @Payload 校验配置一个 Validator。更多信息请参阅@KafkaListener @Payload 校验。

现在你可以直接在注解上指定 kafka 消费者属性；这些属性将覆盖消费者工厂中定义的同名属性（自 2.2.4 版本起）。更多信息请参阅注解属性。

头映射变更​

类型为 MimeType 和 MediaType 的头现在在 RecordHeader 值中被映射为简单字符串。此前它们被映射为 JSON，且只有 MimeType 能被解码。MediaType 无法被解码。现在为了互操作性，它们是简单字符串。

此外，JsonKafkaHeaderMapper 新增了一个 addToStringClasses 方法，允许指定应使用 toString() 而非 JSON 来映射的类型。更多信息请参阅消息头。

嵌入式 Kafka 变更​

KafkaEmbedded 类及其 KafkaRule 接口已弃用，取而代之的是 EmbeddedKafkaBroker 及其 JUnit 4 的 EmbeddedKafkaRule 包装器。@EmbeddedKafka 注解现在会填充一个 EmbeddedKafkaBroker bean，而不是已弃用的 KafkaEmbedded。这一变更允许在 JUnit 5 测试中使用 @EmbeddedKafka。@EmbeddedKafka 注解现在有 ports 属性来指定填充 EmbeddedKafkaBroker 的端口。更多信息请参阅测试应用程序。

JsonSerializer/Deserializer 增强​

现在你可以使用生产者和消费者属性提供类型映射信息。

反序列化器上提供了新的构造器，允许用所提供的目标类型覆盖类型头信息。

JsonDeserializer 现在默认会移除所有类型信息头。

现在你可以使用一个 Kafka 属性来配置 JsonDeserializer 忽略类型信息头（自 2.2.3 版本起）。

Kafka Streams 变更​

流配置 bean 现在必须是一个 KafkaStreamsConfiguration 对象，而不是 StreamsConfig 对象。

StreamsBuilderFactoryBean 已从包 …​core 移至 …​config。

引入了 KafkaStreamBrancher，以便在 KStream 实例之上构建条件分支时提供更好的最终用户体验。

更多信息请参阅Apache Kafka Streams 支持和配置。

Transactional ID​

当事务由监听器容器启动时，transactional.id 现在是 transactionIdPrefix 后接 <group.id>.<topic>.<partition>。这一变更允许对僵尸进行正确隔离，如此处所述。

2.0 与 2.1 之间的变更​

Kafka 客户端版本​

此版本需要 1.0.0 或更高版本的 kafka-clients。

2.2 版本原生支持 1.1.x 客户端。

JSON 改进​

StringJsonMessageConverter 和 JsonSerializer 现在会在 Headers 中添加类型信息，让转换器和 JsonDeserializer 在接收时根据消息本身而非固定的已配置类型创建具体类型。更多信息请参阅序列化、反序列化与消息转换。

停止容器的错误处理器​

现在为记录和批监听器都提供了容器错误处理器，它们将监听器抛出的任何异常都视为致命异常。它们会停止容器。更多信息请参阅处理异常。

暂停与恢复容器​

监听器容器现在有了 pause() 和 resume() 方法（自 2.1.3 版本起）。更多信息请参阅暂停与恢复监听器容器。

有状态重试​

从 2.1.3 版本开始，你可以配置有状态重试。更多信息请参阅有状态重试。

客户端 ID​

从 2.1.1 版本开始，现在你可以在 @KafkaListener 上设置 client.id 前缀。此前，要定制客户端 ID，你需要为每个监听器单独配置消费者工厂（和容器工厂）。此前缀后接 -n，以便在使用并发时提供唯一的客户端 ID。

偏移量提交日志​

默认情况下，主题偏移量提交的日志以 DEBUG 日志级别记录。从 2.1.2 版本开始，ContainerProperties 中一个名为 commitLogLevel 的新属性让你可以为这些消息指定日志级别。更多信息请参阅使用 KafkaMessageListenerContainer。

默认 @KafkaHandler​

从 2.1.3 版本开始，你可以将类级 @KafkaListener 上的某个 @KafkaHandler 注解指定为默认。更多信息请参阅类上的 @KafkaListener。

ReplyingKafkaTemplate​

从 2.1.3 版本开始，提供了 KafkaTemplate 的一个子类以支持请求/回复语义。更多信息请参阅使用 ReplyingKafkaTemplate。

ChainedKafkaTransactionManager​

2.1.3 版本引入了 ChainedKafkaTransactionManager。（它现已弃用。）

从 2.0 迁移指南​

请参阅 2.0 到 2.1 迁移指南。

1.3 与 2.0 之间的变更​

Spring Framework 与 Java 版本​

Spring for Apache Kafka 项目现在要求 Spring Framework 5.0 和 Java 8。

@KafkaListener 变更​

现在你可以用 @SendTo 注解 @KafkaListener 方法（以及类和 @KafkaHandler 方法）。如果方法返回结果，它会被转发到指定主题。更多信息请参阅使用 @SendTo 转发监听器结果。

消息监听器​

消息监听器现在可以感知 Consumer 对象。更多信息请参阅消息监听器。

使用 ConsumerAwareRebalanceListener​

重平衡监听器现在可以在重平衡通知期间访问 Consumer 对象。更多信息请参阅重平衡监听器。

1.2 与 1.3 之间的变更​

对事务的支持​

0.11.0.0 客户端库新增了对事务的支持。新增了 KafkaTransactionManager 及其他事务支持。更多信息请参阅事务。

对头的支持​

0.11.0.0 客户端库新增了对消息头的支持。现在可以在 spring-messaging MessageHeaders 之间进行映射。更多信息请参阅消息头。

创建主题​

0.11.0.0 客户端库提供了一个 AdminClient，你可以用它创建主题。KafkaAdmin 使用此客户端自动添加定义为 @Bean 实例的主题。

对 Kafka 时间戳的支持​

KafkaTemplate 现在支持一个 API，用于添加带时间戳的记录。引入了关于 timestamp 支持的新 KafkaHeaders。此外，还新增了 KafkaConditions.timestamp() 和 KafkaMatchers.hasTimestamp() 测试工具。更多信息请参阅使用 KafkaTemplate、@KafkaListener 注解和测试应用程序。

@KafkaListener 变更​

现在你可以配置一个 KafkaListenerErrorHandler 来处理异常。更多信息请参阅处理异常。

默认情况下，@KafkaListener 的 id 属性现在用作 group.id 属性，覆盖消费者工厂中配置的属性（如果存在）。此外，你可以在注解上显式配置 groupId。此前，你需要单独的容器工厂（和消费者工厂）才能为监听器使用不同的 group.id 值。要恢复使用工厂配置的 group.id 的此前行为，请将注解上的 idIsGroup 属性设为 false。

@EmbeddedKafka 注解​

为方便起见，提供了一个测试类级别的 @EmbeddedKafka 注解，用于将 KafkaEmbedded 注册为 bean。更多信息请参阅测试应用程序。

Kerberos 配置​

现在提供了对配置 Kerberos 的支持。更多信息请参阅 JAAS 与 Kerberos。

1.1 与 1.2 之间的变更​

此版本使用 0.10.2.x 客户端。

1.0 与 1.1 之间的变更​

Kafka 客户端​

此版本使用 Apache Kafka 0.10.x.x 客户端。

批监听器​

监听器可以配置为接收 consumer.poll() 操作返回的整个批消息，而不是一次一条。

空负载​

当你使用日志压缩时，空负载用于"删除"键。

初始偏移量​

显式分配分区时，现在你可以为消费者组配置相对于当前位置的初始偏移量，而不是绝对的或相对于当前末尾的偏移量。

Seek​

现在你可以 seek 每个主题或分区的位置。当使用组管理且 Kafka 分配分区时，你可以用它在初始化期间设置初始位置。你也可以在检测到空闲容器时，或在应用程序执行的任意时刻进行 seek。更多信息请参阅Seek 到特定偏移量。