another fix

databricks · Apr 9, 2024 · 48b0595 · 48b0595
1 parent 13f015e
commit 48b0595
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java b/kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java
@@ -149,7 +149,7 @@ private Map<TopicPartition, Long> committedOffsets(AdminFactory adminFactory, St
   }
 
   private void throwExceptionIfCoordinatorIsTerminated() {
-    if (isLeader && coordinatorThread.isTerminated()) {
+    if (isLeader && coordinatorThread != null && coordinatorThread.isTerminated()) {
       throw new IllegalStateException("Coordinator unexpectedly terminated");
     }
   }
@@ -245,7 +245,7 @@ private <T> void close(T closeable) {
   @Override
   public void close() throws IOException {
     close(commitRequestListener);
-    if (isLeader) {
+    if (isLeader && coordinatorThread != null) {
       coordinatorThread.terminate();
     }
     close(producer);