kafka connect: coordinator only commits offsets when greater than existing offsets

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

@@ -21,6 +21,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
 import org.apache.iceberg.connect.IcebergSinkConfig;
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
   }
 
   protected void commitConsumerOffsets() {
+    Set<TopicPartition> partitions =
+        controlTopicOffsets().keySet().stream()
+            .map(k -> new TopicPartition(controlTopic, k))
+            .collect(Collectors.toSet());
+    Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(partitions);
+
     Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
     controlTopicOffsets()
         .forEach(
-            (k, v) ->
-                offsetsToCommit.put(new TopicPartition(controlTopic, k), new OffsetAndMetadata(v)));
-    consumer.commitSync(offsetsToCommit);
+            (partition, offsetToCommit) -> {
+              TopicPartition tp = new TopicPartition(controlTopic, partition);
+              OffsetAndMetadata lastCommitted = committed.get(tp);
+              if (lastCommitted == null || offsetToCommit > lastCommitted.offset()) {
+                offsetsToCommit.put(tp, new OffsetAndMetadata(offsetToCommit));
+              }
+            });
+    if (!offsetsToCommit.isEmpty()) {
+      LOG.info("Coordinator committing offsets: {}", offsetsToCommit);
+      consumer.commitSync(offsetsToCommit);
+    } else {
+      LOG.info(
+          "Skipping consumer offset commit; local offsets {} are not ahead of committed offsets {}",
+          controlTopicOffsets(),
+          committed);
+    }
   }
 
   void start() {

kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java

@@ -28,6 +28,7 @@
 
 import java.time.OffsetDateTime;
 import java.util.List;
+import java.util.Set;
 import java.util.UUID;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
@@ -53,9 +54,11 @@
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.sink.SinkTaskContext;
 import org.junit.jupiter.api.Test;
@@ -316,6 +319,35 @@ private void triggerCommitCycle(Coordinator coordinator) {
     coordinator.process();
   }
 
+  @Test
+  public void commitConsumerOffsetsShouldNotCommitLowerOffset() {
+    when(config.commitIntervalMs()).thenReturn(0);
+    when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+    SinkTaskContext context = mock(SinkTaskContext.class);
+    Coordinator coordinator =
+        new Coordinator(catalog, config, ImmutableList.of(), clientFactory, context);
+    coordinator.start();
+    initConsumer();
+
+    TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+    long healthyWatermark = 100L;
+    consumer.commitSync(ImmutableMap.of(ctl, new OffsetAndMetadata(healthyWatermark)));
+
+    coordinator.controlTopicOffsets().put(0, 5L);
+    coordinator.commitConsumerOffsets();
+
+    long committed =
+        consumer.committed(Set.of(ctl)).get(ctl) == null
+            ? 0L
+            : consumer.committed(Set.of(ctl)).get(ctl).offset();
+
+    assertThat(committed)
+        .as("commitConsumerOffsets should not rewind the shared -coord consumer group offsets")
+        .isGreaterThanOrEqualTo(healthyWatermark);
+  }
+
   private void assertCommitTable(int idx, UUID commitId, OffsetDateTime ts) {
     byte[] bytes = producer.history().get(idx).value();
     Event commitTable = AvroUtil.decode(bytes);