Kafka Connect: Add otel tracing for kafka iceberg sink

[dttung2905]

Summary

This fixes : #17115
Adds optional OpenTelemetry distributed tracing to the Iceberg Kafka Connect sink. When enabled, the connector creates spans for the two main pipeline stages and continues upstream traces by extracting the W3C traceparent header from Kafka Connect record headers.

Tracing is opt-in (iceberg.tracing.enabled=false by default), uses opentelemetry-api as compileOnly, and degrades gracefully when the API is missing.

Closes #<issue-number> (link your GitHub issue here)

Motivation

CDC pipelines commonly flow Source → Debezium → Kafka → Iceberg sink. Debezium already injects traceparent into record headers (ActivateTracingSpan SMT, debezium-connector-cassandra#173), but the Iceberg sink did not consume that context — traces broke at the sink boundary.

This PR positions the Iceberg sink as the downstream consumer that continues distributed traces and exposes Iceberg-specific latency (ingest vs commit).

Note: This is span tracing for the Connect sink pipeline. It is distinct from core OTEL metrics (#16169) and JMX metrics for the commit pipeline (#17025).

---

Architecture

sequenceDiagram
    participant Upstream as Upstream (Debezium / any traceparent injector)
    participant Kafka as Kafka Topic
    participant SW as SinkWriter
    participant CO as Coordinator

    Upstream->>Kafka: record + traceparent header
    Kafka->>SW: SinkRecord
    SW->>SW: extract traceparent from headers
    SW->>SW: span iceberg-sink-ingest (buffer)
    Note over SW,CO: default 5 min commit interval
    CO->>CO: span iceberg-sink-commit (durability)
    CO->>CO: Iceberg catalog commit

Loading

Spans

Span	Stage	Scope	Parent context
iceberg-sink-ingest	Buffer	Per record written to connector file buffers (Parquet/ORC)	Extracted from SinkRecord.headers() (traceparent)
iceberg-sink-commit	Durability	Per coordinator commit cycle	New root span (intentionally decoupled in v1)

Package layout

Class	Role
Tracing	Facade — centralizes tracing.enabled && OTEL available gate, one-time misconfiguration warning
TracingUtils	Span creation, attribute setting, exception recording
KafkaConnectHeadersGetter	TextMapGetter<Headers> for W3C context extraction from Connect headers

Instrumentation points

• SinkWriter.writeToTable() — ingest span after table routing is resolved (correct db.table, supports static/dynamic routing)
• Coordinator.commit() — commit span wrapping doCommit()
• CommitterImpl.initialize() — one-time warning if tracing enabled but API missing

Caveats and limitations

Possible double spans with OTEL Java agent

If the Connect worker runs the OTEL Java agent with Kafka Connect SinkTask instrumentation enabled, operators may see both:

• Agent batch-level SinkTask.put spans
• Connector per-record iceberg-sink-ingest spans

Ingest span ≠ catalog durability

iceberg-sink-ingest ends when the record is buffered to a file, not when it is committed to the Iceberg catalog. Do not use ingest span duration as end-to-end write latency.

Commit spans are not linked to ingest spans

There is no span link or shared trace ID between a record's ingest span and the commit that makes it durable. Correlating them requires matching on kafka.topic/kafka.offset or future span-link work.

Multi-table routing creates multiple ingest spans

When static routing sends one record to multiple tables, one iceberg-sink-ingest span is created per target table (sibling spans under the same upstream parent).

Debezium envelope attributes are best-effort

op and ts_ms are extracted only when the record value is a Connect Struct with those fields. Plain JSON/Avro records won't have them.

isOpenTelemetryAvailable() means API on classpath, not export configured

GlobalOpenTelemetry.get() succeeds with a no-op tracer when only the API JAR is present. Actual trace export requires SDK + exporter or the OTEL Java agent on the worker JVM.

Out of scope (this PR)

• OTEL metrics (counters/histograms)
• Kafka Connect ConnectMetrics / JMX registration
• Span links between buffered writes and async commits
• OTEL messaging semconv (SpanKind.CONSUMER, messaging.destination.name) — uses INTERNAL + custom attributes for Debezium compatibility

Related work

• Prior work: debezium-connector-cassandra#173
• Core OTEL metrics (different scope): #16169
• Connect JMX metrics (complementary): #17025

TODO:

• : Update docs
• : Document some decision points