Spring Kafka从2.8.x升级到3.0.9后自定义recordInterceptor失效咨询

问题原因 #

Spring Kafka 3.0.x 版本对earlyRecordInterceptor的初始化逻辑做了调整：

• 2.8.x 版本：当isInterceptBeforeTx()为true 或 无事务管理器时，初始化early拦截器
• 3.0.9 版本：仅当isInterceptBeforeTx()为true 且 存在事务管理器时，才会初始化early拦截器

该变更旨在避免拦截器返回null时跳过ErrorHandler的处理流程。

是否需要迁移至ErrorHandler？ #

不需要。ErrorHandler的设计目标是处理消息消费过程中抛出的异常，而你的场景是提前过滤无需处理的消息，用ErrorHandler实现过滤属于场景错配，并非最优选择。

可行解决方案 #

根据你的业务场景，推荐以下几种方案：

方案1：启用事务并开启interceptBeforeTx #

若你的消费流程可引入事务，只需两步配置：

1. 配置Kafka事务管理器
2. 在容器配置中设置interceptBeforeTx=true

此时earlyRecordInterceptor会正常初始化，拦截器即可像2.8.x版本一样生效。注意：拦截器中手动调用consumer.commitAsync()可能与事务的offset提交逻辑冲突，建议移除手动提交，交由Spring Kafka事务机制统一管理offset。

方案2：改用FilteringMessageListenerAdapter #

Spring Kafka提供了专门的消息过滤适配器FilteringMessageListenerAdapter，比自定义RecordInterceptor更适配过滤场景：

class NotSupportedMessageFilter {
    private val consumerTokens = setOf<String>("sample")

    fun getFilterStrategy(): RecordFilterStrategy<Any, Any> {
        return RecordFilterStrategy { record ->
            val headerIterator = record.headers()
                .headers(DefaultJackson2JavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME).iterator()
            if (!headerIterator.hasNext()) {
                return@RecordFilterStrategy false
            }
            val token = String(headerIterator.next().value(), StandardCharsets.UTF_8)
            consumerTokens.contains(token) // 返回true保留消息，false过滤消息
        }
    }
}

// 在配置类中包装原Listener
@Bean
fun filteringListener(originalListener: YourMessageListener): MessageListener<Any, Any> {
    val filter = NotSupportedMessageFilter().getFilterStrategy()
    return FilteringMessageListenerAdapter(originalListener, filter)
}

这种方式无需依赖拦截器的初始化逻辑变更，专门针对消息过滤场景设计，稳定性更强。

方案3：调整拦截器逻辑适配新调用时机 #

若不想引入事务或更换适配器，拦截器仍会在事务后、消息处理前的阶段被调用（late拦截）。此时需修改offset提交逻辑：

• 移除consumer.commitAsync()，返回null后由Spring Kafka根据ackMode配置自动提交offset
• 若容器ackMode为MANUAL或MANUAL_IMMEDIATE，需在过滤后手动提交offset（但late阶段提交需注意逻辑适配）

不过这种方式不如前两种方案直接，不推荐作为首选。

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