dttung2905 opened a new pull request, #17117: URL: https://github.com/apache/iceberg/pull/17117
## Summary 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](https://debezium.io/documentation/reference/stable/integrations/tracing.html), [debezium-connector-cassandra#173](https://github.com/debezium/debezium-connector-cassandra/pull/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](https://github.com/apache/iceberg/issues/16169)) and JMX **metrics** for the commit pipeline ([#17025](https://github.com/apache/iceberg/pull/17025)). --- ## Architecture ```mermaid 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 ``` ### 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](https://github.com/debezium/debezium-connector-cassandra/pull/173) - Core OTEL metrics (different scope): [#16169](https://github.com/apache/iceberg/issues/16169) - Connect JMX metrics (complementary): [#17025](https://github.com/apache/iceberg/pull/17025) ## TODO: - [ ] : Update docs - [ ] : Document some decision points -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
