Kafka Connect: Require every expected partition before completing a commit

[nahidupa]

Part of the #16282 investigation. Not a complete fix for it.

Problem

CommitState decided readiness by comparing a total against the number of partitions the coordinator was constructed with:

receivedPartitionCount += dataComplete.assignments().size();
...
if (receivedPartitionCount >= expectedPartitionCount)

Cardinality does not establish coverage. Three inputs raise the total without covering the expected assignment:

• a DataComplete redelivered by a control-topic replay
• two workers transiently claiming the same source partition while an assignment moves
• a partition no worker owns any more, or one on a different topic

Any of them can satisfy the check while a partition that never reported is still missing. The coordinator then performs a full commit — a subset of the cycle's data, stamped with a kafka.connect.valid-through-ts the table does not satisfy. That watermark is a completeness claim consumers act on.

sequenceDiagram
    autonumber
    participant W0 as Worker, src-topic-0
    participant W1 as Worker, src-topic-1
    participant K as Control topic
    participant C as Coordinator
    participant T as Iceberg table

    Note over C: expects {src-topic-0, src-topic-1}
    W0->>K: DataWritten + DataComplete(src-topic-0)
    K-->>C: total = 1
    Note over K,C: replay, or a stale/foreign partition
    K-->>C: total = 2 -- but still only src-topic-0 covered
    Note over W1: src-topic-1 has never reported
    C->>T: full commit, valid-through-ts stamped
    Note over C,T: snapshot holds one partition's data,<br/>watermark asserts both

Loading

After the fix

Successful full-commit path for one existing table, with an unchanged captured assignment and no timeout. All reports below use the current commit ID; all buffered assignment timestamps are non-null. Partition 0 reports the later timestamp (t0 > t1).

sequenceDiagram
    autonumber
    participant W0 as Worker, src-topic-0
    participant W1 as Worker, src-topic-1
    participant K as Control topic
    participant C as Coordinator
    participant T as Iceberg table

    Note over C: Active commit<br/>Expected = {src-topic-0, src-topic-1}
    W0->>K: DataWritten + DataComplete(src-topic-0, t0)
    K-->>C: Deliver partition 0 events
    C->>C: Record src-topic-0<br/>containsAll(expected) = false

    K-->>C: Replay partition 0 events
    C->>C: Reported identities unchanged<br/>containsAll(expected) = false
    Note over C,T: No snapshot, completion event,<br/>or committed-offset advance
    Note over C: An unexpected identity also<br/>cannot cover missing src-topic-1

    W1->>K: DataWritten + DataComplete(src-topic-1, t1)
    K-->>C: Deliver partition 1 events
    C->>C: Record src-topic-1<br/>containsAll(expected) = true
    C->>C: validThroughTs = min(t0, t1) = t1
    C->>T: Commit buffered files<br/>with control offsets and valid-through-ts = t1
    T-->>C: Snapshot committed
    C->>K: CommitToTable(commitId, snapshotId, t1)
    C->>C: Checkpoint control-topic consumer offsets<br/>and clear buffered file responses
    C->>K: CommitComplete(commitId, t1)
    C->>C: End current commit<br/>and clear readiness state

Loading

Replayed events still reach the coordinator; this fix prevents them from substituting for the missing expected partition. Readiness covers the captured assignment only; assignment freshness, timeout-driven partial commits, and stale-ID timestamp filtering remain separate concerns.

Change

Coordinator retains the expected identities instead of their count, and readiness requires coverage of that set:

// Coordinator
this.expectedPartitions =
    members.stream()
        .flatMap(member -> member.assignment().topicPartitions().stream())
        .collect(Collectors.toUnmodifiableSet());

// CommitState
if (reportedPartitions.containsAll(expectedPartitions))

Reported identities are keyed on (topic, partition), so a matching partition number on a different topic does not count.

flowchart LR
    A[DataComplete arrives] --> B{commit id matches?}
    B -- no --> Z[no partition recorded]
    B -- yes --> C["record (topic, partition)"]
    C --> D{"reported ⊇ expected?"}
    D -- no --> E[keep waiting]
    D -- yes --> F[full commit]

    classDef ok fill:#dcfce7,stroke:#16a34a,color:#14532d
    class C,F ok

Loading

Commit-id filtering is unchanged. A DataComplete whose id does not match contributes no identity, but it still enters readyBuffer. Its timestamps can therefore lower validThroughTs, or suppress it entirely — hasValidThroughTs requires allMatch(timestamp != null) across the whole buffer, so one null from a stale payload yields null. Neither direction can raise the watermark. Excluding stale-commit entries from the validThroughTs calculation was proposed in #17080, which was closed unmerged; that calculation change leaves insertion into readyBuffer unchanged and is a separate concern from readiness, not addressed here.

Scope

This checks coverage against the assignment captured when the coordinator was constructed. Establishing that the captured assignment is still current — a topic expansion while a leader is retained, for instance — is a separate concern and is not addressed here. The watermark is truthful for the captured assignment, not unconditionally.

Relationship to the other control-topic work

An earlier revision of this PR carried a skip-before-dispatch guard in Channel.consumeAvailable. That guard has been removed and the branch rebuilt on current main.

PR	Concern	Status
#17933	keep controlTopicOffsets monotonic	merged
#17713	skip already-consumed control records before dispatch	open; this PR defers to it
#17925 (this)	readiness must cover the expected assignment	the remaining delta
#18006	replacement-coordinator recovery via earliest	separate

Channel.java is untouched, so the merged TestChannel and its hasSize(7) expectation are unchanged here.

The fixes are complementary. Skipping only helps when a replay is behind the tracked position; a replay that is not behind is dispatched normally. Neither offset monotonicity nor dispatch skipping prevents an unexpected or foreign partition from satisfying a count.

Deliberate change to existing expectations

TestCommitState reused a single mocked TopicPartitionOffset for several assignments and asserted against integer counts. Those tests encoded the behaviour this change removes, so they now use distinct identities and expected sets. This is called out here rather than left for a reviewer to find.

Tests

TestCommitState

• testReplayedReadyDoesNotSatisfyQuorumTwice — a redelivered response does not complete the quorum; the commit completes once the missing partition reports.
• testOverlappingAssignmentsDoNotSatisfyQuorumTwice — two workers claiming one partition cover one partition; a second topic pins that the key is (topic, partition).
• unexpectedPartitionsDoNotSatisfyReadiness — an unexpected identity does not substitute for a missing expected one.
• testIsCommitReadyIgnoresZombieCoordinatorPayloads — the stale payload now carries a partition the current commit never reports, so a broken commit-id guard fails the assertion.
• readinessRequiresActiveCommit, emptyAssignmentIsReadyDuringCommit — pin the existing behaviour at the boundaries.

TestCoordinator

• repeatedDataCompleteWaitsForEveryExpectedPartition — parameterised over a same-offset rewind (consumer.seek) and duplicate payloads at new offsets. Asserts no snapshot, no completion event and no checkpoint advance while a partition is missing; then exactly one snapshot with one added file, VALID_THROUGH_TS equal to the earlier of the two timestamps, the expected kafka.connect.offsets, and the committed consumer offset.
• unexpectedPartitionDoesNotCompleteCommit — an unexpected topic does not complete the commit at coordinator level.

Negative controls

Each was applied to production code, run, and reverted:

Mutation	Result
commit-id guard removed	zombie test fails
topic dropped from the recorded key	cross-topic test fails
cardinality readiness restored	both unexpected-partition tests fail; coordinator emits completion early
duplicate-accepting counting restored	the replay unit test and both coordinator replay variants fail; both coordinator variants produce a premature snapshot carrying the later timestamp, while the unit test fails its readiness assertion

Source hashes were captured before and after and confirmed identical, so no mutation was left behind.

Validation

./gradlew -DsparkVersions= -DflinkVersions= -DkafkaVersions=3 \
  :iceberg-kafka-connect:iceberg-kafka-connect:check -x integrationTest --rerun-tasks

BUILD SUCCESSFUL — 19 suites, 149 tests, zero failures, errors or skips. Spotless and both Checkstyle tasks pass. git diff --check clean.

Deterministic unit tests with an in-memory Iceberg catalog and Kafka mock clients. No live-broker rebalance was exercised.

Out of scope

The only cross-restart deduplication floor is the kafka.connect.offsets summary that lastCommittedOffsetsForTable finds by walking snapshot ancestry. Custom summary properties are not propagated by SnapshotProducer, so once every snapshot carrying it has expired the floor is gone. That is independent of this change and of #18006.

---

AI Disclosure

• Model: Claude Opus 5; GPT-6 Astra.
• Platform/Tool: opencode; GitHub Copilot.
• Human Oversight: partially reviewed.
• Prompt Summary: AI assistance was used for codebase analysis, the readiness change, regression tests, negative controls, validation runs, and this description. Review feedback drove three revisions: the original reset-on-rebalance proposal was withdrawn after it was shown to risk data loss; the skip-before-dispatch guard was withdrawn once merge(..., Long::max) merged and the overlap with Kafka Connect: Fix coordinator potentially committing files from prior commit in certain rebalance scenarios #17713 was identified; and count-based readiness was replaced with expected-set coverage after review showed cardinality alone does not establish it. Test claims were checked by mutating production code and confirming the intended tests fail.

[wombatu-kun]

The second paragraph is implementation rationale rather than the method's contract, and That is what we want uses a personal pronoun, which AGENTS.md rules out in comments. Drop it - the reasoning belongs in the PR description.

[nahidupa]

The reset() method was removed when the reset-on-rebalance approach was withdrawn, so this Javadoc and the personal pronoun are no longer present. The remaining implementation rationale is a comment beside addReady; the method contract is not being used to explain the design history.

[wombatu-kun]

Dropping commitBuffer assumes every discarded DataWritten is re-read, but createConsumer leaves auto.offset.reset at latest and the -coord group only gets a committed offset inside doCommit, so a revoke before that group's first successful commit rewinds to the log end instead. Those files already had their source offsets committed by the worker's transaction - is that window intentional?

[nahidupa]

Thanks @wombatu-kun. Your review opened my eyes to a recovery assumption I had missed. That data-loss window was not intentional, and my earlier statement that discarded events would always be re-read was incorrect. With no committed coordinator offset and latest, clearing the buffer can lose the only remaining announcement for files whose source offsets the worker has already committed.

The revised direction is to separate two problems:

• Replay within the same coordinator instance: preserve the in-flight buffers and skip already-consumed control records per partition in consumeAvailable, before updating offsets or dispatching events. This avoids replay-induced readiness double-counting and offset regression without discarding pending files. It overlaps with #17713, so consolidation makes sense.
• Recovery after coordinator replacement: [New BUG Discovered] #18006 changes the coordinator’s default to earliest when control-topic offsets are missing or out of range, while retaining the worker default of latest and respecting explicit overrides. The recovery tests check actual table contents, including committed files mixed with pending files and a subsequent replay. Please have a look at this PR too.

Your feedback also prompted a deeper investigation that uncovered additional problems. Found a duplicate file registration when replay occurs after the snapshot containing the connector’s offset boundary has expired, even though the files remain live. This also helped identify a startup-memory risk from buffering large retained histories before commit-time filtering. Those need separate follow-ups; earliest alone does not make every recovery scenario safe.

Would you recommend consolidating the same-instance replay fix into #17713 and keeping #18006 focused on replacement recovery? I’d appreciate any better approach you see, particularly for preserving pending files while keeping replay deduplication reliable across restarts and snapshot expiration.

[wombatu-kun]

Consolidate, and rebase first: #17933 merged merge(..., Long::max) into this same loop yesterday, so the offset half is already on main and the only delta left is skipping dispatch. Its merged TestChannel asserts seven envelopes dispatched after a two-offset replay, so whichever PR carries the skip has to change a committer-approved expectation on purpose - lead with the DataComplete double-count in addReady, which the merge does not fix. On the durable side, the only cross-restart floor is the kafka.connect.offsets summary that lastCommittedOffsetsForTable finds by walking snapshot ancestry, so expiring that snapshot drops it - separate from both #18006 and this guard.

[nahidupa]

Following up on both the original recovery concern and your later consolidation/snapshot-expiration comment: 2e9d2e927 is rebuilt on main and removes both reset() and the dispatch guard. This PR no longer introduces buffer discard on rebalance; normal successful-commit cleanup is unchanged. Channel.java is untouched relative to the new base.

With the default latest and no valid coordinator checkpoint, a replacement can still skip retained announcements buffered by its predecessor. #18006 addresses that reset default separately; it is not solved by the readiness change here.

Agreed on the durable boundary. If no reachable ancestor retains the per-table connector offset summary, the lookup returns an empty map; ordinary snapshots do not automatically propagate that custom property. Replaying retained announcements after that boundary is lost can register an already-committed file again. That limitation is documented on #18006 and remains outside both changes; earliest alone does not solve it.

The remaining readiness scope and the consolidation question are summarized here: #17925 (comment)

[wombatu-kun]

No test method under org.apache.iceberg.connect carries a Javadoc block; the house style here is a one-to-three line // comment placed at the step it explains. Trim this to the line that says what the test drives and leave the mechanism to the PR description.

[nahidupa]

The reset-era test containing this Javadoc was removed in the rebuild. The replacement readiness tests use short comments at the relevant steps rather than test-method Javadocs.

[wombatu-kun]

Both new tests assert only that nothing was committed early, so nothing covers that the replayed DataWritten still reaches the table exactly once after the rebalance. Deliver both source partitions' DataComplete for the new commit id and assert the file lands in a single snapshot.

[nahidupa]

The new testReplayedDataCompleteStillCommitsTheFileExactlyOnce adds the positive assertion: with two source partitions, duplicate DataWritten/DataComplete payloads from partition 0 do not create a snapshot; after partition 1 reports, the table has one snapshot and its added file locations contain exactly the expected file.

There is an important limit to that coverage: the duplicate payloads are delivered at newer Kafka offsets, not by rewinding and reconsuming the original offsets. The test therefore establishes one file addition within this modeled commit, not the complete rebalance/recovery guarantee requested here. A same-offset rewind case and non-null watermark assertions remain coverage gaps. The current test does fail at the early-snapshot assertion when partition deduplication is disabled.

[wombatu-kun]

#17713 fixes the same eager-rebalance replay from the other side - it skips already-consumed control records in consumeAvailable instead of discarding the in-flight commit - and @uros-b has an open consolidation question there, since #17376 and #17450 carry a byte-equivalent guard and #17933 makes controlTopicOffsets monotonic. Worth stating in the description how this relates: as far as I can tell the reset is the only one of the four that also clears a coordinator's buffer after it loses its partitions, which a skip guard leaves intact.

[wombatu-kun]

main now merges this offset with Long::max in the same loop, so the monotonicity half of the guard has already landed and only the skip-before-dispatch is new here. Rebase onto that and lead with the replayed DataComplete double-count in CommitState.addReady, which the merge does not address.

[nahidupa]

Rebuilt on main at f05bf491a as 2e9d2e927. The only production delta is now in CommitState: count distinct source topic/partition identities for the active commit instead of summing assignment entries. Channel.java, including the merged Long::max handling, is unchanged.

This leads with the double-counting invariant in addReady and removes the duplicated dispatch guard from this PR. It complements dispatch protection rather than replacing it; neither this change nor its tests establish that all rebalance or recovery cases are safe.

[wombatu-kun]

A TestChannel already exists at this path on main, and its controlTopicOffsetsTrackTheHighestPositionConsumed asserts seven dispatched envelopes after replaying two already-consumed offsets - this guard makes that five. Fold these cases into that class and change the assertion deliberately, since it currently encodes the opposite expectation.

[nahidupa]

The duplicated dispatch guard and its proposed TestChannel changes were removed. The rebuilt PR leaves both Channel.java and the merged TestChannel unchanged, including the hasSize(7) dispatch expectation. Any intentional change to that expectation belongs with the PR carrying the guard.

[wombatu-kun]

No record is re-delivered here - rebalance clears the mock's buffer and the LATEST reset lands the position at 2, where the earlier poll already left it - so this never reaches the guard. Was the intent to pin buffer retention rather than the skip, and is that still worth a test now that no rebalance listener exists?

[nahidupa]

Correct: that test did not exercise the dispatch guard. It was removed when the PR was narrowed to readiness counting. The replacement coordinator test delivers duplicate payloads explicitly and fails without partition deduplication, but it uses newer offsets rather than a same-offset rewind. I have called out that remaining coverage gap rather than treating it as a rebalance test.

[wombatu-kun]

dataWrittenEvent in this class builds exactly this event, and this test already calls it for the second file. Use it here and in retainsBufferedFilesWhenRebalanceResetsToLatest.

[nahidupa]

Both referenced tests were removed in the rebuild, along with the helper added for them. The current coordinator regression constructs one DataWritten event inline, so the repeated construction identified in the earlier diff is no longer present.

[wombatu-kun]

Both partial-replay cases pass with the guard removed: the replay is exactly one offset behind, so the unguarded put rewrites the same value, and distinctByKey collapses the duplicate envelope. Consume two records on the replayed partition before the reassignment and replay only the first, so the offsets map would actually regress and the snapshot-offsets assertion catches it.

[nahidupa]

Agreed that the old offset arrangement did not distinguish the guard from the unguarded path. Those cases were removed, and this PR no longer changes offset tracking or dispatch.

The new readiness tests have been checked with a negative control: replacing the reported-partition set with a duplicate-accepting list fails exactly testReplayedReadyDoesNotSatisfyQuorumTwice, testOverlappingAssignmentsDoNotSatisfyQuorumTwice, and testReplayedDataCompleteStillCommitsTheFileExactlyOnce; the other 21 focused tests pass. Restoring the set passes all 24. That demonstrates sensitivity to partition double-counting, not coverage of an actual Kafka rewind. The full validation results and limitations are recorded in the summary comment.

[nahidupa]

Follow-up on the negative-control results above. Those applied to the previous revision, where readiness compared reportedPartitions.size() against an expected count — replacing the set with a duplicate-accepting list inflated that size and produced the reported failures. That evidence was valid for that implementation.

In 65320ff8f readiness requires reportedPartitions.containsAll(expectedPartitions). Duplicates do not change a membership check, so the same Set-to-List mutation would now pass and is no longer a discriminating control. The set retains efficient lookup and avoids storing duplicate identities, but it is containsAll that establishes readiness.

Controls re-run against the current implementation: restoring cardinality readiness fails both unexpected-partition tests; restoring duplicate-accepting counting fails the replay unit test and both coordinator replay variants (the unit test fails its readiness assertion; both coordinator variants produce premature snapshots carrying the later timestamp); removing the commit-id guard fails the zombie test; dropping topic identity fails the cross-topic test. All mutations were reverted and source hashes confirmed identical. Full connector check: 149 tests, zero failures.

One clarification to the description: a stale-id DataComplete still enters readyBuffer, so it can lower validThroughTs or suppress it entirely, since hasValidThroughTs requires a non-null timestamp across the whole buffer. Excluding stale-commit entries from the validThroughTs calculation was proposed in #17080, which was closed unmerged. That change filters the calculation, not insertion into readyBuffer; treating it as a separate concern here.

[nahidupa]

@wombatu-kun Following up on the consolidation comment: the branch has been rebuilt on main (f05bf491a) as 2e9d2e927. Both the reset-on-rebalance approach and the skip-before-dispatch guard have been removed. Channel.java and the merged TestChannel are unchanged relative to that base.

The remaining change is confined to readiness: each source topic/partition contributes at most once for the active commit ID. Long::max keeps offsets monotonic, but does not prevent repeated DataComplete assignments from inflating readiness. A dispatch guard and this set-based check overlap for ordinary record replay, but enforce different invariants: one avoids repeated dispatch, while the other also handles separate responses that report the same partition. This is not a replacement for all of the protection in #17713.

I independently rechecked the tests at this head. Temporarily replacing the partition set with a duplicate-accepting list fails exactly the three new regressions out of 24 focused tests; restoring the set passes all 24. A fresh full connector unit run passed 144 tests across 19 suites, and the connector check gates passed. That corrects the 143-test count currently in the description.

One qualification to the current test wording: the coordinator regression delivers duplicate payloads at newer Kafka offsets. It checks that no early snapshot is created and that, after the missing partition reports, the file is added once in one snapshot. It does not yet exercise a same-offset consumer rewind, and its null timestamps do not verify a non-null completeness watermark. No real-broker rebalance was tested. This change prevents duplicate partition contributions; it does not establish that the expected partition count is current or solve coordinator-replacement recovery and durable replay deduplication.

Would you prefer this readiness change to remain here, or be included alongside #17713? The duplicated dispatch guard is no longer proposed in this PR.

[wombatu-kun]

The set makes coverage checkable, but isCommitReady still compares cardinality against Coordinator's summed totalPartitionCount, so a partition no worker owns any more can still stand in for one that never reported. Was passing the expected Set<TopicPartition> down and checking containsAll considered here?

[nahidupa]

Implemented in 65320ff. Coordinator now captures an immutable Set<TopicPartition> from the member assignments, and CommitState.isCommitReady requires reportedPartitions.containsAll(expectedPartitions). An unexpected partition can no longer substitute for a missing expected one.

Added unexpectedPartitionsDoNotSatisfyReadiness and unexpectedPartitionDoesNotCompleteCommit; both fail when readiness is changed back to a cardinality comparison. This checks coverage against the assignment captured at coordinator construction, not its freshness after a rebalance or topic expansion.

[wombatu-kun]

Both new tests drive addReady with the same observable input - two DataCompletes each yielding ("src-topic", 0), since production reads only topic() and partition(). Give the overlapping case a second topic so it also pins that the key is (topic, partition) and not the partition number.

[nahidupa]

Addressed in 65320ff. testOverlappingAssignmentsDoNotSatisfyQuorumTwice now expects src-topic/0 and other-topic/0: two reports for src-topic/0 remain insufficient, and a report for other-topic/0 completes that expected set. It also checks that an additional expected src-topic/1 is still missing.

The test fails when topic identity is dropped from the recorded key. That gives it a distinct assertion beyond the duplicate-response case.

[wombatu-kun]

With the zombie payload and the current payload both carrying partition 0, this test now passes with the commit-id guard in addReady deleted. Give the zombie payload partitions the current one does not report, so a broken guard still fails the isCommitReady(2) assertion.

[nahidupa]

Addressed in 65320ff. The stale-ID payload now carries partition 1 while the current-ID payload carries partition 0. Readiness for both partitions stays false until partition 1 is reported with the current commit ID.

Removing the commit-ID guard makes testIsCommitReadyIgnoresZombieCoordinatorPayloads fail. The guard was restored after the negative-control run.

[wombatu-kun]

Everything from the config stubs down to initConsumer repeats startCoordinator, which differs only in passing ImmutableList.of() for the members. Add a Collection<MemberDescription> overload to startCoordinator and call it here.

[nahidupa]

Addressed in 65320ff. Added startCoordinator(Collection<MemberDescription> members) and made the no-argument helper delegate with ImmutableList.of(). The assignment-aware tests now use the overload, sharing the config stubs, coordinator startup, and consumer initialization.

[wombatu-kun]

Both DataComplete payloads carry a null timestamp, so validThroughTs stays null and no kafka.connect.valid-through-ts reaches the snapshot. Give partition 0 a later timestamp than partition 1 and assert VALID_THROUGH_TS_SNAPSHOT_PROP the way testCommitAppend does.

[nahidupa]

Addressed in 65320ff. Partition 0 now reports a timestamp one second later than partition 1. repeatedDataCompleteWaitsForEveryExpectedPartition asserts that no snapshot, completion event, or checkpoint advance occurs while partition 1 is missing, then checks that VALID_THROUGH_TS_SNAPSHOT_PROP and both completion-event timestamps equal partition 1's earlier timestamp.

This runs for both a same-offset rewind and duplicate payloads at new offsets. Restoring duplicate-accepting counting makes both coordinator variants fail with premature snapshots carrying the later timestamp. All mutations were reverted; the forced connector check passed 149 tests, with no live-broker rebalance exercised.

[wombatu-kun]

Keep it here - the diff no longer overlaps #17713, which touches only Channel.java and its own test, so folding it in would tie a self-contained CommitState fix to that PR's open consolidation question. The one to reconcile with instead is #17080: it edits CommitState.java and TestCommitState.java to filter readyBuffer by commit id inside validThroughTs, it was closed by the stale bot without ever being reviewed, and issue #17077 is still open.