From ad5bbfd367e188c64c995dd89b2646444616f7ea Mon Sep 17 00:00:00 2001 From: dttung2905 Date: Mon, 6 Jul 2026 22:40:11 +0100 Subject: [PATCH 1/2] Add otel tracing for kafka iceberg sink Signed-off-by: dttung2905 --- kafka-connect/build.gradle | 2 + .../iceberg/connect/IcebergSinkConfig.java | 12 ++ .../connect/channel/CommitterImpl.java | 2 + .../iceberg/connect/channel/Coordinator.java | 10 +- .../iceberg/connect/data/SinkWriter.java | 12 +- .../tracing/KafkaConnectHeadersGetter.java | 62 ++++++++ .../iceberg/connect/tracing/Tracing.java | 76 +++++++++ .../iceberg/connect/tracing/TracingUtils.java | 144 ++++++++++++++++++ .../connect/TestIcebergSinkConfig.java | 13 ++ .../TestKafkaConnectHeadersGetter.java | 55 +++++++ .../iceberg/connect/tracing/TestTracing.java | 72 +++++++++ .../connect/tracing/TestTracingUtils.java | 88 +++++++++++ 12 files changed, 544 insertions(+), 4 deletions(-) create mode 100644 kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java create mode 100644 kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java create mode 100644 kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java create mode 100644 kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java create mode 100644 kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java create mode 100644 kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java diff --git a/kafka-connect/build.gradle b/kafka-connect/build.gradle index d84634dec573..7dba2fea8dd0 100644 --- a/kafka-connect/build.gradle +++ b/kafka-connect/build.gradle @@ -52,7 +52,9 @@ project(':iceberg-kafka-connect:iceberg-kafka-connect') { compileOnly libs.kafka.clients compileOnly libs.kafka.connect.api compileOnly libs.kafka.connect.json + compileOnly 'io.opentelemetry:opentelemetry-api:1.62.0' + testImplementation 'io.opentelemetry:opentelemetry-api:1.62.0' testImplementation libs.hadoop3.client testRuntimeOnly project(':iceberg-parquet') testRuntimeOnly project(':iceberg-orc') diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java index 9fcb130e4094..d52f606f5cff 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java @@ -90,6 +90,7 @@ public class IcebergSinkConfig extends AbstractConfig { private static final String TRANSACTIONAL_PREFIX_PROP = "iceberg.coordinator.transactional.prefix"; private static final String HADOOP_CONF_DIR_PROP = "iceberg.hadoop-conf-dir"; + public static final String TRACING_ENABLED_PROP = "iceberg.tracing.enabled"; private static final String NAME_PROP = "name"; private static final String TASK_ID = "task.id"; @@ -235,6 +236,13 @@ private static ConfigDef newConfigDef() { 120000L, Importance.LOW, "config to control coordinator executor keep alive time"); + configDef.define( + TRACING_ENABLED_PROP, + ConfigDef.Type.BOOLEAN, + false, + Importance.LOW, + "Enables OpenTelemetry tracing. When enabled and the OpenTelemetry API is on the " + + "classpath, the connector creates tracing spans for record ingest and commits."); return configDef; } @@ -448,6 +456,10 @@ public boolean schemaCaseInsensitive() { return getBoolean(TABLES_SCHEMA_CASE_INSENSITIVE_PROP); } + public boolean tracingEnabled() { + return getBoolean(TRACING_ENABLED_PROP); + } + public JsonConverter jsonConverter() { return jsonConverter; } diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java index 7b2d4a25363d..0b49e1b36774 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java @@ -25,6 +25,7 @@ import org.apache.iceberg.connect.Committer; import org.apache.iceberg.connect.IcebergSinkConfig; import org.apache.iceberg.connect.data.SinkWriter; +import org.apache.iceberg.connect.tracing.Tracing; import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.admin.ConsumerGroupDescription; @@ -59,6 +60,7 @@ private void initialize( this.context = sinkTaskContext; this.clientFactory = new KafkaClientFactory(config.kafkaProps()); this.taskId = config.connectorName() + "-" + config.taskId(); + Tracing.ensureConfigured(config); } } diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java index 517fe5225a97..957390042c95 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java @@ -48,6 +48,7 @@ import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.connect.IcebergSinkConfig; +import org.apache.iceberg.connect.tracing.Tracing; import org.apache.iceberg.connect.events.CommitComplete; import org.apache.iceberg.connect.events.CommitToTable; import org.apache.iceberg.connect.events.DataWritten; @@ -151,7 +152,14 @@ protected boolean receive(Envelope envelope) { private void commit(boolean partialCommit) { try { - doCommit(partialCommit); + Map> commitMap = commitState.tableCommitMap(); + Tracing.traceCommit( + config, + commitState.currentCommitId().toString(), + partialCommit, + commitMap.size(), + config.connectGroupId(), + () -> doCommit(partialCommit)); } catch (RuntimeException e) { if (partialCommit) { partialCommitFailures.incrementAndGet(); diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java index 48a01881935b..3b6814c0a4c5 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java @@ -29,6 +29,7 @@ import java.util.stream.Collectors; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.connect.IcebergSinkConfig; +import org.apache.iceberg.connect.tracing.Tracing; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.kafka.common.TopicPartition; @@ -98,7 +99,7 @@ private void routeRecordStatically(SinkRecord record) { .tables() .forEach( tableName -> { - writerForTable(tableName, record, false).write(record); + writeToTable(tableName, record, false); }); } else { @@ -110,7 +111,7 @@ private void routeRecordStatically(SinkRecord record) { tableName -> { Pattern regex = config.tableConfig(tableName).routeRegex(); if (regex != null && regex.matcher(routeValue).matches()) { - writerForTable(tableName, record, false).write(record); + writeToTable(tableName, record, false); } }); } @@ -124,10 +125,15 @@ private void routeRecordDynamically(SinkRecord record) { String routeValue = extractRouteValue(record.value(), routeField); if (routeValue != null) { String tableName = routeValue.toLowerCase(Locale.ROOT); - writerForTable(tableName, record, true).write(record); + writeToTable(tableName, record, true); } } + private void writeToTable(String tableName, SinkRecord record, boolean ignoreMissingTable) { + RecordWriter writer = writerForTable(tableName, record, ignoreMissingTable); + Tracing.traceRecordIngest(config, record, tableName, () -> writer.write(record)); + } + private String extractRouteValue(Object recordValue, String routeField) { if (recordValue == null) { return null; diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java new file mode 100644 index 000000000000..aef9f707ac68 --- /dev/null +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.connect.tracing; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.apache.kafka.connect.header.Header; +import org.apache.kafka.connect.header.Headers; +import io.opentelemetry.context.propagation.TextMapGetter; + +public enum KafkaConnectHeadersGetter implements TextMapGetter { + INSTANCE; + + @Override + public Iterable keys(Headers headers) { + if (headers == null) { + return Collections.emptyList(); + } + List keys = new ArrayList<>(); + for (Header header : headers) { + keys.add(header.key()); + } + return keys; + } + + @Override + public String get(Headers headers, String key) { + if (headers == null) { + return null; + } + Header header = headers.lastWithName(key); + if (header == null) { + return null; + } + Object value = header.value(); + if (value == null) { + return null; + } + if (value instanceof byte[]) { + return new String((byte[]) value, StandardCharsets.UTF_8); + } + return value.toString(); + } +} diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java new file mode 100644 index 000000000000..9110ce6b3eb9 --- /dev/null +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.connect.tracing; + +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.iceberg.connect.IcebergSinkConfig; +import org.apache.kafka.connect.sink.SinkRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Central entry point for opt-in OpenTelemetry tracing in the Kafka Connect sink. */ +public final class Tracing { + + private static final Logger LOG = LoggerFactory.getLogger(Tracing.class); + private static final AtomicBoolean WARNED_MISSING_API = new AtomicBoolean(false); + + private Tracing() {} + + public static boolean isActive(IcebergSinkConfig config) { + return config.tracingEnabled() && TracingUtils.isOpenTelemetryAvailable(); + } + + /** + * Logs a one-time warning when tracing is enabled in config but the OpenTelemetry API is not on + * the classpath. + */ + public static void ensureConfigured(IcebergSinkConfig config) { + if (config.tracingEnabled() + && !TracingUtils.isOpenTelemetryAvailable() + && WARNED_MISSING_API.compareAndSet(false, true)) { + LOG.warn( + "iceberg.tracing.enabled=true but OpenTelemetry API is not on the classpath. " + + "Tracing will be skipped."); + } + } + + public static void traceRecordIngest( + IcebergSinkConfig config, SinkRecord record, String tableName, Runnable ingestAction) { + if (isActive(config)) { + TracingUtils.traceRecordIngest(record, tableName, ingestAction); + } else { + ingestAction.run(); + } + } + + public static void traceCommit( + IcebergSinkConfig config, + String commitId, + boolean partialCommit, + int tableCount, + String connectGroupId, + Runnable commitAction) { + if (isActive(config)) { + TracingUtils.traceCommit( + commitId, partialCommit, tableCount, connectGroupId, commitAction); + } else { + commitAction.run(); + } + } +} diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java new file mode 100644 index 000000000000..6faa6caa2be5 --- /dev/null +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.connect.tracing; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.context.propagation.TextMapPropagator; +import org.apache.kafka.connect.data.Struct; +import org.apache.kafka.connect.sink.SinkRecord; + +public final class TracingUtils { + + private static final String TRACING_COMPONENT = "iceberg-kafka-connect"; + private static final String SINK_INGEST = "iceberg-sink-ingest"; + private static final String SINK_COMMIT = "iceberg-sink-commit"; + private static final String STAGE_BUFFER = "buffer"; + private static final String STAGE_DURABILITY = "durability"; + + private static final boolean OPEN_TELEMETRY_AVAILABLE = resolveOpenTelemetryApiAvailable(); + private static final OpenTelemetry OPEN_TELEMETRY = + OPEN_TELEMETRY_AVAILABLE ? GlobalOpenTelemetry.get() : null; + private static final Tracer TRACER = + OPEN_TELEMETRY_AVAILABLE ? OPEN_TELEMETRY.getTracer(TRACING_COMPONENT) : null; + + private TracingUtils() {} + + public static boolean isOpenTelemetryAvailable() { + return OPEN_TELEMETRY_AVAILABLE; + } + + /** + * Creates a child span for ingesting a record into connector file buffers. Extracts the upstream + * trace context from Kafka Connect record headers (e.g. {@code traceparent} injected by Debezium). + * + *

This span covers buffering into Parquet/ORC files via {@code RecordWriter.write()}, not the + * later Iceberg catalog commit. + */ + public static void traceRecordIngest(SinkRecord record, String tableName, Runnable ingestAction) { + TextMapPropagator propagator = OPEN_TELEMETRY.getPropagators().getTextMapPropagator(); + Context parent = + propagator.extract(Context.root(), record.headers(), KafkaConnectHeadersGetter.INSTANCE); + + Span span = + TRACER.spanBuilder(SINK_INGEST) + .setParent(parent) + .setSpanKind(SpanKind.INTERNAL) + .startSpan(); + try (Scope ignored = span.makeCurrent()) { + span.setAttribute("iceberg.stage", STAGE_BUFFER); + span.setAttribute("db.table", tableName); + span.setAttribute("kafka.topic", record.topic()); + span.setAttribute("kafka.partition", record.kafkaPartition()); + span.setAttribute("kafka.offset", record.kafkaOffset()); + setDebeziumAttributes(span, record); + runAction(span, ingestAction); + } finally { + span.end(); + } + } + + /** Creates a root span for an Iceberg catalog commit cycle (durability). */ + public static void traceCommit( + String commitId, + boolean partialCommit, + int tableCount, + String connectGroupId, + Runnable commitAction) { + Span span = + TRACER.spanBuilder(SINK_COMMIT).setSpanKind(SpanKind.INTERNAL).startSpan(); + try (Scope ignored = span.makeCurrent()) { + span.setAttribute("iceberg.stage", STAGE_DURABILITY); + span.setAttribute("iceberg.commit-id", commitId); + span.setAttribute("iceberg.tables", tableCount); + span.setAttribute("iceberg.partial-commit", partialCommit); + span.setAttribute("kafka.connect.group-id", connectGroupId); + runAction(span, commitAction); + } finally { + span.end(); + } + } + + private static void runAction(Span span, Runnable action) { + try { + action.run(); + } catch (Throwable t) { + span.recordException(t); + String message = t.getMessage(); + span.setStatus(StatusCode.ERROR, message != null ? message : t.getClass().getName()); + throw t; + } + } + + private static void setDebeziumAttributes(Span span, SinkRecord record) { + Object value = record.value(); + if (!(value instanceof Struct)) { + return; + } + Struct struct = (Struct) value; + if (struct.schema().field("op") != null) { + Object op = struct.get("op"); + if (op != null) { + span.setAttribute("op", op.toString()); + } + } + if (struct.schema().field("ts_ms") != null) { + Object tsMs = struct.get("ts_ms"); + if (tsMs instanceof Number) { + span.setAttribute("ts_ms", ((Number) tsMs).longValue()); + } + } + } + + private static boolean resolveOpenTelemetryApiAvailable() { + try { + GlobalOpenTelemetry.get(); + return true; + } catch (NoClassDefFoundError e) { + // ignored + } + return false; + } +} diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java index cc211da797fb..78aa1e65f80d 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java @@ -57,6 +57,19 @@ public void testGetDefault() { "iceberg.tables", "db.landing"); IcebergSinkConfig config = new IcebergSinkConfig(props); assertThat(config.commitIntervalMs()).isEqualTo(300_000); + assertThat(config.tracingEnabled()).isFalse(); + } + + @Test + public void testTracingEnabled() { + Map props = + ImmutableMap.of( + "iceberg.catalog.type", "rest", + "topics", "source-topic", + "iceberg.tables", "db.landing", + IcebergSinkConfig.TRACING_ENABLED_PROP, "true"); + IcebergSinkConfig config = new IcebergSinkConfig(props); + assertThat(config.tracingEnabled()).isTrue(); } @Test diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java new file mode 100644 index 000000000000..19de7b3396e1 --- /dev/null +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.connect.tracing; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.header.ConnectHeaders; +import org.apache.kafka.connect.header.Headers; +import org.junit.jupiter.api.Test; + +public class TestKafkaConnectHeadersGetter { + + private static final String TRACEPARENT = + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + + @Test + public void testGetStringHeader() { + Headers headers = new ConnectHeaders(); + headers.addString("traceparent", TRACEPARENT); + + assertThat(KafkaConnectHeadersGetter.INSTANCE.get(headers, "traceparent")).isEqualTo(TRACEPARENT); + assertThat(KafkaConnectHeadersGetter.INSTANCE.keys(headers)).contains("traceparent"); + } + + @Test + public void testGetStringHeaderViaSchema() { + Headers headers = new ConnectHeaders(); + headers.add("traceparent", TRACEPARENT, Schema.STRING_SCHEMA); + + assertThat(KafkaConnectHeadersGetter.INSTANCE.get(headers, "traceparent")).isEqualTo(TRACEPARENT); + } + + @Test + public void testNullHeadersReturnsEmptyKeys() { + assertThat(KafkaConnectHeadersGetter.INSTANCE.keys(null)).isEmpty(); + assertThat(KafkaConnectHeadersGetter.INSTANCE.get(null, "traceparent")).isNull(); + } +} diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java new file mode 100644 index 000000000000..4265f60cc4aa --- /dev/null +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.connect.tracing; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.google.common.collect.ImmutableMap; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.iceberg.connect.IcebergSinkConfig; +import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.jupiter.api.Test; + +public class TestTracing { + + private static IcebergSinkConfig config(boolean tracingEnabled) { + return new IcebergSinkConfig( + ImmutableMap.of( + "iceberg.catalog.type", + "rest", + "topics", + "source-topic", + "iceberg.tables", + "db.landing", + IcebergSinkConfig.TRACING_ENABLED_PROP, + Boolean.toString(tracingEnabled))); + } + + @Test + public void testIsActiveWhenTracingDisabled() { + assertThat(Tracing.isActive(config(false))).isFalse(); + } + + @Test + public void testIsActiveWhenTracingEnabled() { + assertThat(Tracing.isActive(config(true))).isTrue(); + } + + @Test + public void testTraceRecordIngestSkipsWhenDisabled() { + AtomicBoolean ingested = new AtomicBoolean(false); + SinkRecord record = new SinkRecord("topic", 0, null, null, null, null, 1L); + + Tracing.traceRecordIngest(config(false), record, "db.table", () -> ingested.set(true)); + + assertThat(ingested).isTrue(); + } + + @Test + public void testTraceCommitSkipsWhenDisabled() { + AtomicBoolean committed = new AtomicBoolean(false); + + Tracing.traceCommit(config(false), "commit-1", false, 1, "group", () -> committed.set(true)); + + assertThat(committed).isTrue(); + } +} diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java new file mode 100644 index 000000000000..3c78521974b7 --- /dev/null +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.connect.tracing; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.google.common.collect.ImmutableMap; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.iceberg.connect.IcebergSinkConfig; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.SchemaBuilder; +import org.apache.kafka.connect.data.Struct; +import org.apache.kafka.connect.errors.DataException; +import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.jupiter.api.Test; + +public class TestTracingUtils { + + @Test + public void testOpenTelemetryAvailableOnTestClasspath() { + assertThat(TracingUtils.isOpenTelemetryAvailable()).isTrue(); + } + + @Test + public void testTraceRecordIngestCompletesWithNoOpTracer() { + AtomicBoolean ingested = new AtomicBoolean(false); + Schema valueSchema = + SchemaBuilder.struct().field("op", Schema.STRING_SCHEMA).field("ts_ms", Schema.INT64_SCHEMA); + Struct value = new Struct(valueSchema).put("op", "c").put("ts_ms", 12345L); + SinkRecord record = new SinkRecord("topic", 0, null, null, valueSchema, value, 42L); + + TracingUtils.traceRecordIngest(record, "db.table", () -> ingested.set(true)); + + assertThat(ingested).isTrue(); + } + + @Test + public void testTraceRecordIngestPropagatesException() { + SinkRecord record = new SinkRecord("topic", 0, null, null, null, null, 42L); + + assertThatThrownBy( + () -> + TracingUtils.traceRecordIngest( + record, "db.table", () -> { throw new DataException("ingest failed"); })) + .isInstanceOf(DataException.class) + .hasMessage("ingest failed"); + } + + @Test + public void testTraceCommitCompletesWithNoOpTracer() { + AtomicBoolean committed = new AtomicBoolean(false); + + TracingUtils.traceCommit("commit-1", false, 2, "connect-group", () -> committed.set(true)); + + assertThat(committed).isTrue(); + } + + @Test + public void testTraceCommitPropagatesException() { + assertThatThrownBy( + () -> + TracingUtils.traceCommit( + "commit-1", + true, + 1, + "connect-group", + () -> { throw new RuntimeException("commit failed"); })) + .isInstanceOf(RuntimeException.class) + .hasMessage("commit failed"); + } +} From 748642a9de9ee872530fd663305000090faa4890 Mon Sep 17 00:00:00 2001 From: dttung2905 Date: Wed, 8 Jul 2026 21:42:19 +0100 Subject: [PATCH 2/2] Fix failed CI Signed-off-by: dttung2905 --- .../iceberg/connect/channel/Coordinator.java | 2 +- .../tracing/KafkaConnectHeadersGetter.java | 6 +++--- .../apache/iceberg/connect/tracing/Tracing.java | 3 +-- .../iceberg/connect/tracing/TracingUtils.java | 9 +++++---- .../iceberg/connect/TestIcebergSinkConfig.java | 12 ++++++++---- .../tracing/TestKafkaConnectHeadersGetter.java | 6 ++++-- .../iceberg/connect/tracing/TestTracing.java | 2 +- .../connect/tracing/TestTracingUtils.java | 16 +++++++++++----- 8 files changed, 34 insertions(+), 22 deletions(-) diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java index 957390042c95..be2660a96642 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java @@ -48,13 +48,13 @@ import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.connect.IcebergSinkConfig; -import org.apache.iceberg.connect.tracing.Tracing; import org.apache.iceberg.connect.events.CommitComplete; import org.apache.iceberg.connect.events.CommitToTable; import org.apache.iceberg.connect.events.DataWritten; import org.apache.iceberg.connect.events.Event; import org.apache.iceberg.connect.events.StartCommit; import org.apache.iceberg.connect.events.TableReference; +import org.apache.iceberg.connect.tracing.Tracing; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Streams; diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java index aef9f707ac68..d338f09ba770 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/KafkaConnectHeadersGetter.java @@ -18,13 +18,13 @@ */ package org.apache.iceberg.connect.tracing; +import io.opentelemetry.context.propagation.TextMapGetter; import java.nio.charset.StandardCharsets; -import java.util.ArrayList; import java.util.Collections; import java.util.List; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.kafka.connect.header.Header; import org.apache.kafka.connect.header.Headers; -import io.opentelemetry.context.propagation.TextMapGetter; public enum KafkaConnectHeadersGetter implements TextMapGetter { INSTANCE; @@ -34,7 +34,7 @@ public Iterable keys(Headers headers) { if (headers == null) { return Collections.emptyList(); } - List keys = new ArrayList<>(); + List keys = Lists.newArrayList(); for (Header header : headers) { keys.add(header.key()); } diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java index 9110ce6b3eb9..6dd2ca6967c3 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java @@ -67,8 +67,7 @@ public static void traceCommit( String connectGroupId, Runnable commitAction) { if (isActive(config)) { - TracingUtils.traceCommit( - commitId, partialCommit, tableCount, connectGroupId, commitAction); + TracingUtils.traceCommit(commitId, partialCommit, tableCount, connectGroupId, commitAction); } else { commitAction.run(); } diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java index 6faa6caa2be5..8a6022c95f16 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/TracingUtils.java @@ -52,7 +52,8 @@ public static boolean isOpenTelemetryAvailable() { /** * Creates a child span for ingesting a record into connector file buffers. Extracts the upstream - * trace context from Kafka Connect record headers (e.g. {@code traceparent} injected by Debezium). + * trace context from Kafka Connect record headers (e.g. {@code traceparent} injected by + * Debezium). * *

This span covers buffering into Parquet/ORC files via {@code RecordWriter.write()}, not the * later Iceberg catalog commit. @@ -63,7 +64,8 @@ public static void traceRecordIngest(SinkRecord record, String tableName, Runnab propagator.extract(Context.root(), record.headers(), KafkaConnectHeadersGetter.INSTANCE); Span span = - TRACER.spanBuilder(SINK_INGEST) + TRACER + .spanBuilder(SINK_INGEST) .setParent(parent) .setSpanKind(SpanKind.INTERNAL) .startSpan(); @@ -87,8 +89,7 @@ public static void traceCommit( int tableCount, String connectGroupId, Runnable commitAction) { - Span span = - TRACER.spanBuilder(SINK_COMMIT).setSpanKind(SpanKind.INTERNAL).startSpan(); + Span span = TRACER.spanBuilder(SINK_COMMIT).setSpanKind(SpanKind.INTERNAL).startSpan(); try (Scope ignored = span.makeCurrent()) { span.setAttribute("iceberg.stage", STAGE_DURABILITY); span.setAttribute("iceberg.commit-id", commitId); diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java index 78aa1e65f80d..b240d5edbd35 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java @@ -64,10 +64,14 @@ public void testGetDefault() { public void testTracingEnabled() { Map props = ImmutableMap.of( - "iceberg.catalog.type", "rest", - "topics", "source-topic", - "iceberg.tables", "db.landing", - IcebergSinkConfig.TRACING_ENABLED_PROP, "true"); + "iceberg.catalog.type", + "rest", + "topics", + "source-topic", + "iceberg.tables", + "db.landing", + IcebergSinkConfig.TRACING_ENABLED_PROP, + "true"); IcebergSinkConfig config = new IcebergSinkConfig(props); assertThat(config.tracingEnabled()).isTrue(); } diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java index 19de7b3396e1..9b0cd9173c8b 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestKafkaConnectHeadersGetter.java @@ -35,7 +35,8 @@ public void testGetStringHeader() { Headers headers = new ConnectHeaders(); headers.addString("traceparent", TRACEPARENT); - assertThat(KafkaConnectHeadersGetter.INSTANCE.get(headers, "traceparent")).isEqualTo(TRACEPARENT); + assertThat(KafkaConnectHeadersGetter.INSTANCE.get(headers, "traceparent")) + .isEqualTo(TRACEPARENT); assertThat(KafkaConnectHeadersGetter.INSTANCE.keys(headers)).contains("traceparent"); } @@ -44,7 +45,8 @@ public void testGetStringHeaderViaSchema() { Headers headers = new ConnectHeaders(); headers.add("traceparent", TRACEPARENT, Schema.STRING_SCHEMA); - assertThat(KafkaConnectHeadersGetter.INSTANCE.get(headers, "traceparent")).isEqualTo(TRACEPARENT); + assertThat(KafkaConnectHeadersGetter.INSTANCE.get(headers, "traceparent")) + .isEqualTo(TRACEPARENT); } @Test diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java index 4265f60cc4aa..cc3147128bf2 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracing.java @@ -20,9 +20,9 @@ import static org.assertj.core.api.Assertions.assertThat; -import com.google.common.collect.ImmutableMap; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.iceberg.connect.IcebergSinkConfig; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.kafka.connect.sink.SinkRecord; import org.junit.jupiter.api.Test; diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java index 3c78521974b7..a3e499309117 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/tracing/TestTracingUtils.java @@ -21,9 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import com.google.common.collect.ImmutableMap; import java.util.concurrent.atomic.AtomicBoolean; -import org.apache.iceberg.connect.IcebergSinkConfig; import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; @@ -42,7 +40,9 @@ public void testOpenTelemetryAvailableOnTestClasspath() { public void testTraceRecordIngestCompletesWithNoOpTracer() { AtomicBoolean ingested = new AtomicBoolean(false); Schema valueSchema = - SchemaBuilder.struct().field("op", Schema.STRING_SCHEMA).field("ts_ms", Schema.INT64_SCHEMA); + SchemaBuilder.struct() + .field("op", Schema.STRING_SCHEMA) + .field("ts_ms", Schema.INT64_SCHEMA); Struct value = new Struct(valueSchema).put("op", "c").put("ts_ms", 12345L); SinkRecord record = new SinkRecord("topic", 0, null, null, valueSchema, value, 42L); @@ -58,7 +58,11 @@ public void testTraceRecordIngestPropagatesException() { assertThatThrownBy( () -> TracingUtils.traceRecordIngest( - record, "db.table", () -> { throw new DataException("ingest failed"); })) + record, + "db.table", + () -> { + throw new DataException("ingest failed"); + })) .isInstanceOf(DataException.class) .hasMessage("ingest failed"); } @@ -81,7 +85,9 @@ public void testTraceCommitPropagatesException() { true, 1, "connect-group", - () -> { throw new RuntimeException("commit failed"); })) + () -> { + throw new RuntimeException("commit failed"); + })) .isInstanceOf(RuntimeException.class) .hasMessage("commit failed"); }