Kafka Connect: Exclude zombie commitId from validThroughTs()

[thswlsqls]

Closes #17077

Summary

• CommitState.addReady() buffers every DATA_COMPLETE event in readyBuffer regardless of commitId, and validThroughTs() used that buffer unfiltered, so a zombie coordinator's stale event from a previous commit could corrupt or null out VALID_THROUGH_TS_SNAPSHOT_PROP.
• validThroughTs() now filters readyBuffer by Objects.equals(currentCommitId, payload.commitId()), the same pattern addReady() already applies to receivedPartitionCount.
• PR Kafka Connect: Make CommitState.isCommitReady() O(1) #16453 (merged) applied this filter to receivedPartitionCount and isCommitReady() but left validThroughTs() unfiltered — this closes that gap.

Testing done

• Added TestCommitState#testGetValidThroughTsIgnoresZombieCoordinatorPayloads, mirroring testIsCommitReadyIgnoresZombieCoordinatorPayloads from PR Kafka Connect: Make CommitState.isCommitReady() O(1) #16453, covering a zombie coordinator's DATA_COMPLETE (different commitId, null timestamp) being excluded from the valid-through calculation.
• Updated TestCommitState#testGetValidThroughTs to stub commitId() on its mocked DataComplete payloads so they match the new filter.
• ./gradlew :iceberg-kafka-connect:iceberg-kafka-connect:check passes — TestCommitState 5/5, full module 123/123, 0 failures.

---

AI Disclosure

• Tool: Claude Code — used to analyze the code, implement the fix, and write the test.

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

[github-actions]

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

[vbhanuchander-lang]

@thswlsqls this was closed by the stale bot on 28 Aug, not on merit — it was never reviewed by
anyone. The bug is still on main and your fix is correct. I verified it end to end so a committer
has less to redo:

The bug is real and current. addReady() buffers every DataComplete unconditionally
(CommitState.java:63), while the very next branch filters by commitId for
receivedPartitionCount (line 68). validThroughTs() then reads that unfiltered buffer, so a
zombie coordinator's stale event still reaches it on main today. Your framing is the convincing
one: #16453 applied this filter to receivedPartitionCount and isCommitReady() and left
validThroughTs() behind, so this is finishing that change rather than a new idea.

Verified against 8ea7d00 (main, 3 Sep):

• the patch applies cleanly with no conflicts
• TestCommitState passes with it, 5 tests
• reverting only CommitState.java and keeping your test fails with
  expected: 2026-09-03T00:19:38.829847Z but was: null — so
  testGetValidThroughTsIgnoresZombieCoordinatorPayloads genuinely guards the change rather than
  passing either way
• the whole iceberg-kafka-connect module is green, 136 tests

The null timestamp in your zombie payload is what makes the test bite: it flips hasValidThroughTs
to false and nulls the whole result, so the stale event does not merely shift the watermark, it
erases kafka.connect.valid-through-ts from the snapshot.

The bot's message says the PR can be revived at any time. @bryanck would you be willing to reopen
it? Happy to answer anything on the verification, but the change is @thswlsqls's and should stay
theirs.