Kafka Connect: Add otel tracing for kafka iceberg sink

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')

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;
   }

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);
     }
   }
 

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

@@ -54,6 +54,7 @@
 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;
@@ -135,7 +136,7 @@
 
   @Override
   protected boolean receive(Envelope envelope) {
     switch (envelope.event().payload().type()) {
       case DATA_WRITTEN:
         commitState.addResponse(envelope);
         return true;
@@ -151,7 +152,14 @@
 
   private void commit(boolean partialCommit) {
     try {
-      doCommit(partialCommit);
+      Map<TableReference, List<Envelope>> commitMap = commitState.tableCommitMap();
+      Tracing.traceCommit(
+          config,
+          commitState.currentCommitId().toString(),
+          partialCommit,
+          commitMap.size(),
+          config.connectGroupId(),
+          () -> doCommit(partialCommit));
     } catch (RuntimeException e) {
       if (partialCommit) {
         partialCommitFailures.incrementAndGet();

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;

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 io.opentelemetry.context.propagation.TextMapGetter;
+import java.nio.charset.StandardCharsets;
+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;
+
+public enum KafkaConnectHeadersGetter implements TextMapGetter<Headers> {
+  INSTANCE;
+
+  @Override
+  public Iterable<String> keys(Headers headers) {
+    if (headers == null) {
+      return Collections.emptyList();
+    }
+    List<String> keys = Lists.newArrayList();
+    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();
+  }
+}

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/tracing/Tracing.java

@@ -0,0 +1,75 @@
+/*
+ * 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();
+    }
+  }
+}