如何为Debezium MySQL连接器传递Outbox表字段至OpenTelemetry追踪？

1. 确认配置有效性 #

• 检查tracing.span.context.field的字段名是否与Outbox表中实际字段完全匹配（大小写敏感），如果字段嵌套在消息的payload结构中，需写成payload.trace_id这类完整路径。
• 通过Kafka Connect的REST API验证配置是否加载：执行GET /connectors/{connector-name}/config，确认该配置项存在且值正确。

2. 验证Outbox消息结构 #

• 查看Kafka主题中的实际消息，确认目标字段存在于指定路径下。可使用命令行工具查看：

  kafka-console-consumer.sh --bootstrap-server <kafka-host>:9092 --topic <outbox-topic> --from-beginning --property print.key=true --property print.value=true
  

• 确保Debezium已将Outbox表的行记录正确解析为包含目标字段的结构化消息。

3. 检查版本兼容性与JavaAgent配置 #

• 确认Debezium版本（建议1.9+）与OpenTelemetry JavaAgent版本（建议1.10+）兼容，版本不匹配可能导致配置失效。
• 检查Docker启动命令中JavaAgent的参数是否正确，未覆盖Debezium追踪配置：

  docker run -d \
    -e KAFKA_CONNECT_BOOTSTRAP_SERVERS=<kafka-host>:9092 \
    -e KAFKA_CONNECT_GROUP_ID=connect-group \
    -e KAFKA_CONNECT_CONFIG_STORAGE_TOPIC=connect-configs \
    -e KAFKA_CONNECT_OFFSET_STORAGE_TOPIC=connect-offsets \
    -e KAFKA_CONNECT_STATUS_STORAGE_TOPIC=connect-statuses \
    -e JAVA_OPTS="-javaagent:/path/to/opentelemetry-javaagent.jar -Dotel.service.name=kafka-connect-debezium" \
    -v /path/to/opentelemetry-javaagent.jar:/path/to/opentelemetry-javaagent.jar \
    confluentinc/cp-kafka-connect-base:latest
  

4. 自定义SMT注入追踪上下文 #

如果内置配置仍不生效，可通过自定义SMT（Single Message Transform）手动提取字段并注入追踪上下文：

• 编写SMT代码，从消息中提取目标字段并设置到OpenTelemetry Span属性中：

  import io.opentelemetry.api.trace.Span;
  import org.apache.kafka.connect.connector.ConnectRecord;
  import org.apache.kafka.connect.transforms.Transformation;
  import org.apache.kafka.connect.data.Struct;
  
  public class InjectOutboxTraceContext implements Transformation<ConnectRecord> {
      @Override
      public ConnectRecord apply(ConnectRecord record) {
          Struct payload = (Struct) record.value();
          String traceId = payload.getString("trace_id");
          if (traceId != null) {
              Span.current().setAttribute("outbox.trace_id", traceId);
          }
          return record;
      }
  
      @Override
      public void configure(java.util.Map<String, ?> configs) {}
  
      @Override
      public void close() {}
  }
  

• 将SMT打包为Jar，挂载到Docker容器的/usr/share/java/kafka-connect-debezium/目录，然后在连接器配置中添加：

  transforms=injectTrace
  transforms.injectTrace.type=com.example.InjectOutboxTraceContext
  

5. 日志排查 #

• 开启DEBUG级日志定位问题，修改容器内的log4j.properties：

  log4j.logger.io.debezium.tracing=DEBUG
  log4j.logger.io.opentelemetry=DEBUG
  

• 重启容器后，查看日志中是否存在字段找不到、追踪上下文初始化失败等报错信息。

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