apache / apache/iceberg

Kafka Connect: validThroughTs() includes stale commitId entries from zombie coordinators

Open Beginner friendly
#17,077 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.