(iceberg) branch main updated: Kafka Connect: Set engine metadata in EnvironmentContext (#17195)

[am...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

amogh-jahagirdar pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new f660519a4e Kafka Connect: Set engine metadata in EnvironmentContext (#17195)
f660519a4e is described below

commit f660519a4e027a30a74c08b9259ce8ce0ae87755
Author: ajreid21 <57...@users.noreply.github.com>
AuthorDate: Tue Jul 21 09:28:56 2026 -0700

    Kafka Connect: Set engine metadata in EnvironmentContext (#17195)
    
    Set Kafka Connect engine name and version during task startup so commits include the Kafka runtime context.
---
 .../apache/iceberg/connect/IcebergSinkTask.java    |  4 ++
 .../iceberg/connect/TestIcebergSinkTask.java       | 67 ++++++++++++++++++++++
 2 files changed, 71 insertions(+)

diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java
index 7783bebd47..471187172e 100644
--- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java
+++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java
@@ -21,10 +21,12 @@ package org.apache.iceberg.connect;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.utils.AppInfoParser;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
 import org.slf4j.Logger;
@@ -45,6 +47,8 @@ public class IcebergSinkTask extends SinkTask {
 
   @Override
   public void start(Map<String, String> props) {
+    EnvironmentContext.put(EnvironmentContext.ENGINE_NAME, "kafka-connect");
+    EnvironmentContext.put(EnvironmentContext.ENGINE_VERSION, AppInfoParser.getVersion());
     this.config = new IcebergSinkConfig(props);
     this.catalog = CatalogUtils.loadCatalog(config);
     this.committer = CommitterFactory.createCommitter(config);
diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkTask.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkTask.java
new file mode 100644
index 0000000000..f1f71fc131
--- /dev/null
+++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkTask.java
@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.apache.iceberg.EnvironmentContext;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.junit.jupiter.api.Test;
+
+class TestIcebergSinkTask {
+
+  @Test
+  void startSetsEnvironmentContext() {
+    String previousName = EnvironmentContext.remove(EnvironmentContext.ENGINE_NAME);
+    String previousVersion = EnvironmentContext.remove(EnvironmentContext.ENGINE_VERSION);
+    IcebergSinkTask task = new IcebergSinkTask();
+
+    try {
+      task.start(taskConfig());
+
+      assertThat(EnvironmentContext.get())
+          .containsEntry(EnvironmentContext.ENGINE_NAME, "kafka-connect")
+          .containsEntry(EnvironmentContext.ENGINE_VERSION, AppInfoParser.getVersion());
+    } finally {
+      task.stop();
+      restoreEnvironmentContext(EnvironmentContext.ENGINE_NAME, previousName);
+      restoreEnvironmentContext(EnvironmentContext.ENGINE_VERSION, previousVersion);
+    }
+  }
+
+  private static Map<String, String> taskConfig() {
+    return ImmutableMap.of(
+        "name", "test-connector",
+        "task.id", "0",
+        "topics", "mytopic",
+        "iceberg.tables", "mytable",
+        "iceberg.catalog.catalog-impl", InMemoryCatalog.class.getName());
+  }
+
+  private static void restoreEnvironmentContext(String key, String value) {
+    if (value != null) {
+      EnvironmentContext.put(key, value);
+    } else {
+      EnvironmentContext.remove(key);
+    }
+  }
+}