You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "thswlsqls (via GitHub)" <gi...@apache.org> on 2026/07/03 22:50:36 UTC
[PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
thswlsqls opened a new pull request, #17080:
URL: https://github.com/apache/iceberg/pull/17080
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 #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 #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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #17080: Kafka Connect: Exclude zombie commitId from validThroughTs()
URL: https://github.com/apache/iceberg/pull/17080
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4133873192
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -60,12 +60,12 @@ void addResponse(Envelope envelope) {
void addReady(Envelope envelope) {
DataComplete dataComplete = (DataComplete) envelope.event().payload();
- readyBuffer.add(dataComplete);
if (!isCommitInProgress()) {
LOG.warn(
"Received commit ready when no commit in progress, this can happen during recovery. Commit ID: {}",
dataComplete.commitId());
} else if (Objects.equals(currentCommitId, dataComplete.commitId())) {
+ readyBuffer.add(dataComplete);
receivedPartitionCount += dataComplete.assignments().size();
}
Review Comment:
Perhaps we should add a log statement when the commit ID doesn't match
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4123804076
##########
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()))
Review Comment:
Rather than duplicating the same filter in `isCommitReady()` I feel we can filter out the old ready events in `addReady()`, then both are working off of the same collection, do you have thoughts on that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub)" <gi...@apache.org>.
thswlsqls commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4129463563
##########
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()))
Review Comment:
@bryanck Thanks for reopening this and for the review. Agreed, I'll move the commit ID check into `addReady()` so only the current commit's `DataComplete` events go into `readyBuffer`, and drop the extra filter from `validThroughTs()`.
This keeps the same behavior: `startNewCommit()` only runs when no commit is in progress, and `endCurrentCommit()` always clears the buffer, so filtering on the way in yields the same set as filtering on read. It also stops zombie events from piling up in the buffer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub) via issues" <is...@iceberg.apache.org>.
thswlsqls commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-6810636153
@bryanck Gentle ping on this one — it's approved, rebased, and all checks are green, so it should be ready to merge whenever you have a moment.
Thanks again for the review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-5363601919
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub)" <gi...@apache.org>.
thswlsqls commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4139674971
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -60,12 +60,12 @@ void addResponse(Envelope envelope) {
void addReady(Envelope envelope) {
DataComplete dataComplete = (DataComplete) envelope.event().payload();
- readyBuffer.add(dataComplete);
if (!isCommitInProgress()) {
LOG.warn(
"Received commit ready when no commit in progress, this can happen during recovery. Commit ID: {}",
dataComplete.commitId());
} else if (Objects.equals(currentCommitId, dataComplete.commitId())) {
+ readyBuffer.add(dataComplete);
receivedPartitionCount += dataComplete.assignments().size();
}
Review Comment:
@bryanck Thanks — agreed, I'll add an `else` branch in `addReady()` that logs the dropped commit ID together with the current one.
I'll use `warn` to match the "no commit in progress" message right above it. Happy to lower it to `info` if you'd prefer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub)" <gi...@apache.org>.
thswlsqls commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-5868103019
@vbhanuchander-lang Thanks for verifying this end to end. No code change is needed here: the patch still merges cleanly into current `main`, and the `iceberg-kafka-connect` module passes with it.
@bryanck I can't reopen this myself since the bot closed it. Could you reopen it when you have a moment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub)" <gi...@apache.org>.
thswlsqls commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4129586933
##########
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()))
Review Comment:
@bryanck Done in f70818843: `addReady()` now only adds a `DataComplete` to `readyBuffer` when it matches the current commit, and `validThroughTs()` reads `readyBuffer` directly again.
`TestCommitState` (5 tests) and the rest of the `iceberg-kafka-connect` module pass. I've also updated the PR description to match.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub)" <gi...@apache.org>.
thswlsqls commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4162860964
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -60,12 +60,12 @@ void addResponse(Envelope envelope) {
void addReady(Envelope envelope) {
DataComplete dataComplete = (DataComplete) envelope.event().payload();
- readyBuffer.add(dataComplete);
if (!isCommitInProgress()) {
LOG.warn(
"Received commit ready when no commit in progress, this can happen during recovery. Commit ID: {}",
dataComplete.commitId());
} else if (Objects.equals(currentCommitId, dataComplete.commitId())) {
+ readyBuffer.add(dataComplete);
receivedPartitionCount += dataComplete.assignments().size();
}
Review Comment:
@bryanck Done in 49e85d15e: `addReady()` now logs a `warn` with both the dropped and the current commit ID when they don't match.
`TestCommitState` and the rest of the `iceberg-kafka-connect` module still pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "vbhanuchander-lang (via GitHub)" <gi...@apache.org>.
vbhanuchander-lang commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-5518397116
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-5447803888
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "github-actions[bot] (via GitHub) via issues" <is...@iceberg.apache.org>.
github-actions[bot] commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-6810499757
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub)" <gi...@apache.org>.
thswlsqls commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-5976505616
@bryanck Thanks for the review! I've rebased onto the latest `main`, which cleared the two jackson-core CVEs ignored in #18346.
The scan now flags one more jackson-core CVE (CVE-2026-68494), which #18358 adds to the same ignore list.
I'll rebase again once that's merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "thswlsqls (via GitHub)" <gi...@apache.org>.
thswlsqls commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-6116782094
@bryanck Rebased again now that #18358 is merged, and the CVE scan passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4146555591
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -60,12 +60,12 @@ void addResponse(Envelope envelope) {
void addReady(Envelope envelope) {
DataComplete dataComplete = (DataComplete) envelope.event().payload();
- readyBuffer.add(dataComplete);
if (!isCommitInProgress()) {
LOG.warn(
"Received commit ready when no commit in progress, this can happen during recovery. Commit ID: {}",
dataComplete.commitId());
} else if (Objects.equals(currentCommitId, dataComplete.commitId())) {
+ readyBuffer.add(dataComplete);
receivedPartitionCount += dataComplete.assignments().size();
}
Review Comment:
I feel `warn` should be OK.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Kafka Connect: Exclude zombie commitId from validThroughTs() [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#issuecomment-5960937064
LGTM, thanks for the contribution @thswlsqls ! Can you rebase to fix the CVE scan?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org