Apache Iceberg version
main @ 49b89a8
Query engine
N/A — Kafka Connect sink, engine-agnostic
Please describe the bug
CommitState.validThroughTs() (kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java line 146) computes the VALID_THROUGH_TS_SNAPSHOT_PROP snapshot property from readyBuffer without filtering by commitId. addReady() (line 61) buffers every DATA_COMPLETE event regardless of commitId, including ones from a zombie coordinator still emitting events for a previous commit. As a result, a zombie coordinator's stale (or null-timestamp) partition offsets can pull the valid-through timestamp into the past or suppress it entirely (null), even though the current commit's data is complete and valid.
PR #16453 ("Kafka Connect: Make CommitState.isCommitReady() O(1)", merged) added the same Objects.equals(currentCommitId, payload.commitId()) filter to receivedPartitionCount and covered it with testIsCommitReadyIgnoresZombieCoordinatorPayloads, but validThroughTs() was left unfiltered — an incomplete fix.
Steps to reproduce
- Start a commit (
startNewCommit()).
- Have a zombie coordinator emit
DATA_COMPLETE with a different commitId (addReady()).
- Have the current coordinator emit its own
DATA_COMPLETE.
- Call
validThroughTs(false) — the zombie's offsets are still included in the min-timestamp calculation.
Additional context
N/A — reproducible via a pure unit test on CommitState, no Docker required.
Apache Iceberg version
main @ 49b89a8
Query engine
N/A — Kafka Connect sink, engine-agnostic
Please describe the bug
CommitState.validThroughTs()(kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.javaline 146) computes theVALID_THROUGH_TS_SNAPSHOT_PROPsnapshot property fromreadyBufferwithout filtering bycommitId.addReady()(line 61) buffers everyDATA_COMPLETEevent regardless ofcommitId, including ones from a zombie coordinator still emitting events for a previous commit. As a result, a zombie coordinator's stale (or null-timestamp) partition offsets can pull the valid-through timestamp into the past or suppress it entirely (null), even though the current commit's data is complete and valid.PR #16453 ("Kafka Connect: Make CommitState.isCommitReady() O(1)", merged) added the same
Objects.equals(currentCommitId, payload.commitId())filter toreceivedPartitionCountand covered it withtestIsCommitReadyIgnoresZombieCoordinatorPayloads, butvalidThroughTs()was left unfiltered — an incomplete fix.Steps to reproduce
startNewCommit()).DATA_COMPLETEwith a differentcommitId(addReady()).DATA_COMPLETE.validThroughTs(false)— the zombie's offsets are still included in the min-timestamp calculation.Additional context
N/A — reproducible via a pure unit test on
CommitState, no Docker required.