Kafka Connect: validThroughTs() includes stale commitId entries from zombie coordinators
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
**Apache Iceberg version**
main @ 49b89a8c5
**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**
1. Start a commit (`startNewCommit()`).
2. Have a zombie coordinator emit `DATA_COMPLETE` with a different `commitId` (`addReady()`).
3. Have the current coordinator emit its own `DATA_COMPLETE`.
4. 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.
Contributor guide
Research direction
Start in kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java, reading addReady() and validThroughTs(). Reproduce the case with a pure CommitState unit test: a zombie coordinator has a different commitId, while the current coordinator reports DATA_COMPLETE. Done means validThroughTs(false) uses only the current commit's offsets and preserves the expected timestamp.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 78/100