Kafka Connect: Terminate commits on coordinator stop (#10814)

Author: bryanck

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

@@ -53,6 +53,7 @@
 import org.apache.iceberg.util.Tasks;
 import org.apache.iceberg.util.ThreadPools;
 import org.apache.kafka.clients.admin.MemberDescription;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkTaskContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -71,6 +72,7 @@ class Coordinator extends Channel {
   private final String snapshotOffsetsProp;
   private final ExecutorService exec;
   private final CommitState commitState;
+  private volatile boolean terminated;
 
   Coordinator(
       Catalog catalog,
@@ -218,6 +220,10 @@ private void commitToTable(
             .filter(distinctByKey(deleteFile -> deleteFile.path().toString()))
             .collect(Collectors.toList());
 
+    if (terminated) {
+      throw new ConnectException("Coordinator is terminated, commit aborted");
+    }
+
     if (dataFiles.isEmpty() && deleteFiles.isEmpty()) {
       LOG.info("Nothing to commit to table {}, skipping", tableIdentifier);
     } else {
@@ -296,19 +302,18 @@ private Map<Integer, Long> lastCommittedOffsetsForTable(Table table, String bran
     return ImmutableMap.of();
   }
 
-  @Override
-  void stop() {
+  void terminate() {
+    this.terminated = true;
+
     exec.shutdownNow();
 
-    // ensure coordinator tasks are shut down, else cause the sink worker to fail
+    // wait for coordinator termination, else cause the sink task to fail
     try {
       if (!exec.awaitTermination(1, TimeUnit.MINUTES)) {
-        throw new RuntimeException("Timed out waiting for coordinator shutdown");
+        throw new ConnectException("Timed out waiting for coordinator shutdown");
       }
     } catch (InterruptedException e) {
-      throw new RuntimeException("Interrupted while waiting for coordinator shutdown", e);
+      throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
     }
-
-    super.stop();
   }
 }

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

@@ -25,7 +25,7 @@ class CoordinatorThread extends Thread {
   private static final Logger LOG = LoggerFactory.getLogger(CoordinatorThread.class);
   private static final String THREAD_NAME = "iceberg-coord";
 
-  private Coordinator coordinator;
+  private final Coordinator coordinator;
   private volatile boolean terminated;
 
   CoordinatorThread(Coordinator coordinator) {
@@ -39,15 +39,15 @@ public void run() {
       coordinator.start();
     } catch (Exception e) {
       LOG.error("Coordinator error during start, exiting thread", e);
-      terminated = true;
+      this.terminated = true;
     }
 
     while (!terminated) {
       try {
         coordinator.process();
       } catch (Exception e) {
         LOG.error("Coordinator error during process, exiting thread", e);
-        terminated = true;
+        this.terminated = true;
       }
     }
 
@@ -56,14 +56,14 @@ public void run() {
     } catch (Exception e) {
       LOG.error("Coordinator error during stop, ignoring", e);
     }
-    coordinator = null;
   }
 
   boolean isTerminated() {
     return terminated;
   }
 
   void terminate() {
-    terminated = true;
+    this.terminated = true;
+    coordinator.terminate();
   }
 }

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

@@ -61,6 +61,7 @@
 import org.apache.iceberg.types.Types.TimestampType;
 import org.apache.iceberg.util.DateTimeUtil;
 import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.errors.ConnectException;
 
 class RecordConverter {
 
@@ -421,7 +422,7 @@ protected LocalDate convertDateValue(Object value) {
       int days = (int) (((Date) value).getTime() / 1000 / 60 / 60 / 24);
       return DateTimeUtil.dateFromDays(days);
     }
-    throw new RuntimeException("Cannot convert date: " + value);
+    throw new ConnectException("Cannot convert date: " + value);
   }
 
   @SuppressWarnings("JavaUtilDate")
@@ -437,7 +438,7 @@ protected LocalTime convertTimeValue(Object value) {
       long millis = ((Date) value).getTime();
       return DateTimeUtil.timeFromMicros(millis * 1000);
     }
-    throw new RuntimeException("Cannot convert time: " + value);
+    throw new ConnectException("Cannot convert time: " + value);
   }
 
   protected Temporal convertTimestampValue(Object value, TimestampType type) {
@@ -461,7 +462,7 @@ private OffsetDateTime convertOffsetDateTime(Object value) {
     } else if (value instanceof Date) {
       return DateTimeUtil.timestamptzFromMicros(((Date) value).getTime() * 1000);
     }
-    throw new RuntimeException(
+    throw new ConnectException(
         "Cannot convert timestamptz: " + value + ", type: " + value.getClass());
   }
 
@@ -489,7 +490,7 @@ private LocalDateTime convertLocalDateTime(Object value) {
     } else if (value instanceof Date) {
       return DateTimeUtil.timestampFromMicros(((Date) value).getTime() * 1000);
     }
-    throw new RuntimeException(
+    throw new ConnectException(
         "Cannot convert timestamp: " + value + ", type: " + value.getClass());
   }