Spring Boot 3中Kafka消费者无法获取生产者TraceID求助

摘要 #

• 期望：消费者与生产者拥有相同的TraceID
• 实际结果：两者的TraceID不同

问题描述 #

我正在为基于Spring Boot 3的Spring Cloud应用启用链路追踪功能，使用micrometer和zipkin作为追踪工具（Sleuth与Spring Boot 3不兼容）。

已启用生产者追踪，且能在ProducerRecord中看到traceParent头。生产者发送负载代码如下：

public void sendNotificationPayload(NotificationPayload payload) {
        Span span = tracer.currentSpan();
        CompletableFuture<SendResult<String, Object>> future = template.send("notification", payload);
        // future.get() will block
        future.whenComplete((res, err) -> {
            if (Objects.isNull(err)) {
                log.info("send message=[{}] with offset=[{}]", payload.toString(), res.getRecordMetadata().offset());
                res.getProducerRecord().headers().forEach(
                        (h) -> log.info("header:{}, {}", h.key(), new String(h.value(), StandardCharsets.UTF_8))
                );
            } else {
                log.error("Unable to send message=[{}] due to: {}", payload.toString(), err.getMessage());
            }
        });
    }

生产者配置：

@Bean
    public KafkaTemplate kafkaTemplate(ProducerFactory producerFactory) {
        KafkaTemplate t = new KafkaTemplate(producerFactory);
        t.setObservationEnabled(true);
        t.setObservationConvention(new KafkaTemplateObservationConvention() {
            @Override
            public KeyValues getLowCardinalityKeyValues(KafkaRecordSenderContext context) {
                return KeyValues.of("topic", context.getDestination(),
                        "id", String.valueOf(context.getRecord().key()));
            }
        });
        return t;
    }

消费者配置：

@Autowired
    private Tracing tracing;

    @Bean
    public KafkaTracing kafkaTracing(Tracing tracing) {
        return KafkaTracing.newBuilder(tracing).build();
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        // 创建字符串-对象映射配置
        Map<String, Object> config = new HashMap<>();
        return new DefaultKafkaConsumerFactory<>(config);
    }

    // 参考方案：Kafka Tracing with Spring Boot and Open Telemetry
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> listenerFactory(ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.getContainerProperties().setObservationEnabled(true);
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

生产者日志明确显示ProducerRecord包含traceParent头：

2024-06-20T16:43:41.323-05:00  INFO [AUTH-SERVICE,6674a28d99aabe22eb47ae662c67a12d,8655877e8686f26f] 69037 --- [AUTH-SERVICE] [VICE-producer-1] [6674a28d99aabe22eb47ae662c67a12d-8655877e8686f26f] c.j.a.kafka.KafkaMessagePublisher        : header:traceparent, 00-6674a28d99aabe22eb47ae662c67a12d-8655877e8686f26f-01

但消费者始终无法获取到与生产者相同的TraceID。已尝试显式启用生产者和消费者的观测功能，调试确认生产者端traceParent存在，但消费者端TraceID始终不一致，恳请帮忙解决该问题。

参考资料 #

• 同问题社区讨论：Spring Boot社区关于Kafka TraceID传递的疑难问题
• 技术文章：Kafka Tracing with Spring Boot and Open Telemetry
• Stack Overflow问题：跨Kafka事件传递与管理TraceID
• Spring Cloud Stream官方Issue：链路追踪传递相关问题
• 技术博文：Spring Boot 3 + Zipkin + Kafka Transport链路追踪实践

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