diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java
index 922285b13c6d..cf9b68ca03d8 100644
--- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java
+++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java
@@ -144,16 +144,21 @@ Map<TableReference, List<Envelope>> tableCommitMap() {
   }
 
   OffsetDateTime validThroughTs(boolean partialCommit) {
+    List<DataComplete> currentCommitReady =
+        readyBuffer.stream()
+            .filter(payload -> Objects.equals(currentCommitId, payload.commitId()))
+            .collect(Collectors.toList());
+
     boolean hasValidThroughTs =
         !partialCommit
-            && readyBuffer.stream()
+            && currentCommitReady.stream()
                 .flatMap(event -> event.assignments().stream())
                 .allMatch(offset -> offset.timestamp() != null);
 
     OffsetDateTime result;
     if (hasValidThroughTs) {
       result =
-          readyBuffer.stream()
+          currentCommitReady.stream()
               .flatMap(event -> event.assignments().stream())
               .map(TopicPartitionOffset::timestamp)
               .min(Comparator.naturalOrder())
diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitState.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitState.java
index c994a2b9d2e4..8241e1d1a642 100644
--- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitState.java
+++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitState.java
@@ -113,7 +113,11 @@ public void testIsCommitReadyIgnoresZombieCoordinatorPayloads() {
 
   @Test
   public void testGetValidThroughTs() {
+    CommitState commitState = new CommitState(mock(IcebergSinkConfig.class));
+    commitState.startNewCommit();
+
     DataComplete payload1 = mock(DataComplete.class);
+    when(payload1.commitId()).thenReturn(commitState.currentCommitId());
     TopicPartitionOffset tp1 = mock(TopicPartitionOffset.class);
     OffsetDateTime ts1 = EventTestUtil.now();
     when(tp1.timestamp()).thenReturn(ts1);
@@ -124,14 +128,12 @@ public void testGetValidThroughTs() {
     when(payload1.assignments()).thenReturn(ImmutableList.of(tp1, tp2));
 
     DataComplete payload2 = mock(DataComplete.class);
+    when(payload2.commitId()).thenReturn(commitState.currentCommitId());
     TopicPartitionOffset tp3 = mock(TopicPartitionOffset.class);
     OffsetDateTime ts3 = ts1.plusSeconds(2);
     when(tp3.timestamp()).thenReturn(ts3);
     when(payload2.assignments()).thenReturn(ImmutableList.of(tp3));
 
-    CommitState commitState = new CommitState(mock(IcebergSinkConfig.class));
-    commitState.startNewCommit();
-
     commitState.addReady(wrapInEnvelope(payload1));
     commitState.addReady(wrapInEnvelope(payload2));
 
@@ -140,6 +142,7 @@ public void testGetValidThroughTs() {
 
     // null timestamp for one, so should not set a valid-through timestamp
     DataComplete payload3 = mock(DataComplete.class);
+    when(payload3.commitId()).thenReturn(commitState.currentCommitId());
     TopicPartitionOffset tp4 = mock(TopicPartitionOffset.class);
     when(tp4.timestamp()).thenReturn(null);
     when(payload3.assignments()).thenReturn(ImmutableList.of(tp4));
@@ -150,6 +153,33 @@ public void testGetValidThroughTs() {
     assertThat(commitState.validThroughTs(true)).isNull();
   }
 
+  @Test
+  public void testGetValidThroughTsIgnoresZombieCoordinatorPayloads() {
+    CommitState commitState = new CommitState(mock(IcebergSinkConfig.class));
+    commitState.startNewCommit();
+
+    // Stale DataComplete from a zombie Coordinator that started a different commit, with a null
+    // timestamp that would otherwise suppress the valid-through timestamp.
+    DataComplete zombiePayload = mock(DataComplete.class);
+    when(zombiePayload.commitId()).thenReturn(UUID.randomUUID());
+    TopicPartitionOffset zombieTp = mock(TopicPartitionOffset.class);
+    when(zombieTp.timestamp()).thenReturn(null);
+    when(zombiePayload.assignments()).thenReturn(ImmutableList.of(zombieTp));
+
+    DataComplete currentPayload = mock(DataComplete.class);
+    when(currentPayload.commitId()).thenReturn(commitState.currentCommitId());
+    TopicPartitionOffset currentTp = mock(TopicPartitionOffset.class);
+    OffsetDateTime ts = EventTestUtil.now();
+    when(currentTp.timestamp()).thenReturn(ts);
+    when(currentPayload.assignments()).thenReturn(ImmutableList.of(currentTp));
+
+    commitState.addReady(wrapInEnvelope(zombiePayload));
+    commitState.addReady(wrapInEnvelope(currentPayload));
+
+    // Only the current commit's payload counts toward the valid-through timestamp.
+    assertThat(commitState.validThroughTs(false)).isEqualTo(ts);
+  }
+
   private Envelope wrapInEnvelope(Payload payload) {
     Event event = mock(Event.class);
     when(event.payload()).thenReturn(payload);