Spring Boot下如何将Kafka ConsumerRecord Headers透传至ProducerRecord、RestTemplate、Feign

你遇到的ThreadLocal无法清理的问题，可以通过Spring Kafka原生提供的RecordInterceptor的完整生命周期回调解决，Spring Kafka 2.7及以上版本的RecordInterceptor支持afterCompletion方法，会在单条消息消费完成（包含正常结束、异常抛出的所有场景）后触发，完美适配ThreadLocal的设值和清理需求。

方案1：自定义上下文 + RecordInterceptor实现（无额外依赖） #

步骤1：实现自定义Header上下文持有者 #

用ThreadLocal存储透传Header，提供设值、取值、清理方法：

object CustomHeaderContext {
    private val HEADERS_HOLDER = ThreadLocal<Map<String, String>>()

    fun setHeaders(headers: Map<String, String>) {
        HEADERS_HOLDER.set(headers)
    }

    fun getHeaders(): Map<String, String>? {
        return HEADERS_HOLDER.get()
    }

    fun clear() {
        HEADERS_HOLDER.remove()
    }
}

步骤2：实现RecordInterceptor处理Header的存取 #

class KafkaHeaderPropagationInterceptor : RecordInterceptor<Any, Any> {

    override fun intercept(record: ConsumerRecord<Any, Any>, consumer: Consumer<Any, Any>): ConsumerRecord<Any, Any>? {
        // 消费前从ConsumerRecord提取自定义Header存入上下文
        val headers = record.headers().associate { header ->
            header.key() to String(header.value(), Charsets.UTF_8)
        }
        CustomHeaderContext.setHeaders(headers)
        return record
    }

    override fun afterCompletion(
        record: ConsumerRecord<Any, Any>,
        consumer: Consumer<Any, Any>,
        exception: Exception?
    ) {
        // 消费完成后清理上下文，避免线程复用导致的数据错乱
        CustomHeaderContext.clear()
    }
}

步骤3：注册拦截器到Kafka监听容器工厂 #

@Configuration
class KafkaConfig {

    @Bean
    fun kafkaListenerContainerFactory(
        consumerFactory: ConsumerFactory<Any, Any>
    ): ConcurrentKafkaListenerContainerFactory<Any, Any> {
        val factory = ConcurrentKafkaListenerContainerFactory<Any, Any>()
        factory.consumerFactory = consumerFactory
        // 注册自定义Header传播拦截器
        factory.setRecordInterceptor(KafkaHeaderPropagationInterceptor())
        return factory
    }
}

步骤4：兼容原有Feign透传逻辑 #

修改之前的Feign拦截器，优先从RequestContextHolder取HTTP Header，取不到则从Kafka自定义上下文取，实现全链路透传：

class CustomRequestInterceptor : RequestInterceptor {
    override fun apply(template: RequestTemplate) {
        // 先取HTTP请求的Header
        val requestAttributes = RequestContextHolder.getRequestAttributes() as ServletRequestAttributes?
        val headers = requestAttributes?.request?.let { request ->
            request.headerNames.toList().associateWith { request.getHeader(it) }
        } ?: CustomHeaderContext.getHeaders() // 取不到则取Kafka消费的Header

        headers?.forEach { (name, value) ->
            template.header(name, value)
        }
    }
}

方案2：链路追踪组件原生支持（适用已集成Spring Cloud Sleuth/Micrometer Tracing的项目） #

如果你的项目已经接入了链路追踪能力，可以直接将自定义Header配置为链路baggage，组件会自动完成HTTP、Feign、Kafka等全链路的透传，无需手动维护ThreadLocal和拦截器逻辑。

注意事项 #

• 如果消费逻辑中使用了异步线程处理，普通ThreadLocal无法跨线程传递，可替换为TransmittableThreadLocal实现跨线程池的上下文传递
• 仅透传需要的自定义Header即可，不要全量透传所有Kafka Header，避免不必要的性能损耗和数据冗余

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