Kogito Workflow集成OpenTelemetry：异步Kafka步骤Trace上下文统一方法咨询

核心问题根源 #

异步Kafka通信场景下，Kogito默认不会自动传递OpenTelemetry的traceparent上下文到后续步骤，导致每个流程节点生成独立Trace。要实现全链路追踪，核心是在Kafka消息中携带并传递Trace上下文，让Kogito各步骤正确提取并复用该上下文。

具体实现步骤 #

1. 配置Quarkus OpenTelemetry与Kafka集成 #

在application.properties中开启Kafka的OpenTelemetry自动埋点，确保基础Trace功能生效：

quarkus.opentelemetry.enabled=true
quarkus.opentelemetry.tracer.exporter.otlp.endpoint=http://你的Otel采集器地址:4317
# 开启Kafka生产者/消费者的Trace上下文传递
quarkus.opentelemetry.instrumentation.kafka.enabled=true

2. 实现Kafka消息的Trace上下文传递 #

生产者端（Workflow触发下一步时）

发送Kafka消息前，通过OpenTelemetry API提取当前Trace上下文，注入到消息头中：

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.propagation.TextMapPropagator;
import jakarta.inject.Inject;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import java.util.HashMap;
import java.util.Map;

public class StepMessageProducer {

    @Inject
    TextMapPropagator propagator;

    @Outgoing("workflow-step-output")
    public Message<String> sendStepMessage(String payload) {
        // 获取当前Span的上下文
        Context currentContext = Context.current().with(Span.current());
        Map<String, String> headers = new HashMap<>();
        // 将Trace上下文注入消息头
        propagator.inject(currentContext, headers, Map::put);
        
        return Message.of(payload, headers);
    }
}

消费者端（Workflow接收消息时）

消费消息时，从消息头提取Trace上下文，设置为当前上下文，让后续流程步骤复用同一个Trace：

import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import io.opentelemetry.context.propagation.TextMapPropagator;
import jakarta.inject.Inject;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;

public class StepMessageConsumer {

    @Inject
    TextMapPropagator propagator;

    @Incoming("workflow-step-input")
    public void receiveStepMessage(Message<String> message) {
        // 从消息头提取Trace上下文
        Context extractedContext = propagator.extract(
            Context.current(), 
            message.getHeaders(), 
            (carrier, key) -> carrier.get(key)
        );
        
        // 将提取的上下文设为当前上下文，后续操作关联到同一Trace
        try (Scope scope = extractedContext.makeCurrent()) {
            // 执行Workflow步骤逻辑
            handleWorkflowStep(message.getPayload());
        }
    }

    private void handleWorkflowStep(String payload) {
        // 你的业务处理逻辑
    }
}

3. 验证全链路Trace关联 #

• 查看Otel采集器或Jaeger等可视化工具的Trace详情，确认所有流程步骤的Span属于同一个Trace ID
• 检查每个Span的parent.id是否正确指向上游Span，确保链路层级关系正确

4. 关键注意事项 #

• 所有参与流程的服务需使用相同的OpenTelemetry传播器（默认W3CTraceContextPropagator，无需额外配置）
• 若使用Kogito内置消息绑定，需确保Kogito消息组件能识别并传递Trace上下文头
• 流程步骤中避免手动创建根Span，应基于传入的上下文创建子Span

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