Spring Cloud Sleuth Trace ID能否在Kinesis函数式模型中通过Header传递？

我尝试让Spring Cloud Sleuth的Trace ID通过Spring Cloud Stream Kinesis消息传递，但调试测试时发现，除X-B3-TraceId外，其他设置的Header均能在消费者端正常获取。想确认Trace ID传递是否在函数式模型中被支持？

我的配置与代码 #

application.yml配置片段 #

spring:
  cloud:
    stream:
      kinesis:
        binder:
          headers:
            - spring.cloud.function.definition
            - env
            - X-B3-TraceId

消息发布方式 #

streamBridge.send(SERVICE_STREAM, "kinesis", message);

消息构建逻辑 #

protected <T> Message<T> buildMessagefrom(T entity) {
    val clazz = entity.getClass();
    return MessageBuilder
            .withPayload(entity)
            // ...some other headers
            .setHeader("env", "local")
            .setHeader("spring.cloud.function.definition", "myConsumer")
            .setHeader("X-B3-TraceId", "123")
            .build();
}

消费者代码 #

@Bean("myConsumer")
Consumer<Message<AppointmentCreatedEvent>> myConsumer() {
    return message -> {

    };
}

问题解答 #

函数式模型完全支持Spring Cloud Sleuth的Trace ID传递，你的问题大概率是配置或头处理方式的问题，以下是排查和解决方向：

• 不要手动设置X-B3-TraceId头
  Spring Cloud Sleuth会自动将当前上下文的Trace ID注入到消息头中，手动设置会干扰Sleuth的自动传播逻辑。移除buildMessagefrom方法中手动设置X-B3-TraceId的代码，让Sleuth自动处理。

• 调整Kinesis Binder的头配置
  确保配置中包含Sleuth所需的全量B3头，而不仅仅是TraceId。完整的B3头包括X-B3-TraceId、X-B3-SpanId、X-B3-ParentSpanId、X-B3-Sampled，建议直接配置headers: '*'（或者明确列出所有B3头），避免遗漏：

  spring:
    cloud:
      stream:
        kinesis:
          binder:
            headers: '*'
  

  或者更精确的配置：

  spring:
    cloud:
      stream:
        kinesis:
          binder:
            headers:
              - spring.cloud.function.definition
              - env
              - X-B3-TraceId
              - X-B3-SpanId
              - X-B3-ParentSpanId
              - X-B3-Sampled
  

• 确保版本兼容
  检查Spring Cloud Sleuth和Spring Cloud Stream Kinesis Binder的版本是否匹配，建议使用Spring Cloud官方推荐的版本组合（如2022.0.x版本系列），版本不兼容可能导致头传播失效。

• 验证消息头传递
  在消费者端添加日志输出所有头，确认B3头是否存在：

  @Bean("myConsumer")
  Consumer<Message<AppointmentCreatedEvent>> myConsumer() {
      return message -> {
          System.out.println("Received headers: " + message.getHeaders());
      };
  }
  

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