nahidupa opened a new pull request, #17925:
URL: https://github.com/apache/iceberg/pull/17925
Closes #16282
## Problem
The coordinator's control-topic consumer is group-managed via `subscribe()`,
so it participates in rebalances. Nothing observes those rebalances, and
`CommitState` holds the entire in-flight commit in memory:
```java
private final List<Envelope> commitBuffer = Lists.newArrayList();
private final List<DataComplete> readyBuffer = Lists.newArrayList();
private int receivedPartitionCount = 0;
private UUID currentCommitId; // null == no commit in flight
```
When the coordinator loses its assignment mid-commit, the consumer resumes
from its last *committed* offset and re-reads records it has already processed.
The in-flight state survives that rewind, so the replayed events land in a
commit that is still live.
Two things go wrong when they do.
**The readiness quorum is inflated.** `addReady` only skips an event when
the commit ids differ:
```java
} else if (Objects.equals(currentCommitId, dataComplete.commitId())) {
receivedPartitionCount += dataComplete.assignments().size();
}
```
After a rewind the id still matches, so the same `DataComplete` is counted
twice and `isCommitReady()` fires before every partition has reported.
**The recorded offset floor regresses.** `Channel.consumeAvailable` tracks
progress with a plain `put`:
```java
controlTopicOffsets.put(record.partition(), record.offset() + 1);
```
A replay overwrites a higher value with a lower one, while `commitBuffer`
still holds the higher offsets. That map is what `commitToTable` writes into
the snapshot summary as `kafka.connect.offsets.<topic>.<group>`, and it is the
floor the next cycle uses to drop already-committed events:
```java
Long minOffset = committedOffsets.get(envelope.partition());
return minOffset == null || envelope.offset() >= minOffset;
```
So the snapshot ends up advertising a floor behind its own contents, and the
dedup filter is disarmed for exactly the range that was already committed. The
reporter in #16282 observed the end result in production: the same `file_path`
with the same `record_count` registered in two snapshots, and every row in that
file returned twice.
Nothing downstream catches it. `distinctByKey(ContentFile::location)` only
deduplicates within a single batch, and `AppendFiles` has no uniqueness
constraint on data file location, so a duplicate registration is valid metadata.
## Why no `seek()` appears in the code
This was the part that made the bug hard to see. There is no explicit rewind
anywhere. IKC never sets `partition.assignment.strategy`, so the client default
`[RangeAssignor, CooperativeStickyAssignor]` applies and takes its protocol
from the first entry — **eager**. An eager rebalance revokes every partition
and discards its fetch position, even for a member handed the same partition
straight back; `updateFetchPositions()` then refetches the committed offset.
The rewind happens inside the consumer.
## What changes
```mermaid
flowchart LR
R["Partitions revoked"] -->|onPartitionsRevoked|
RS["commitState.reset()"]
R -->|"consumer resumes<br/>from offset 0"| RW["Consumer rewinds"]
RS -->|"no commit id held"| G["Stale event ignored<br/>no commit in
progress"]
RW -->|"replayed DataComplete"| G
G -->|"counter stays at zero"| F["Fresh StartCommit<br/>new id, live
quorum"]
F -->|"refilled in lockstep"| OK["Registered once"]
R -.->|"before this patch"| S["State survives<br/>commit id still held"]
S -.->|"map regresses"| D["Double-counted"]
D -.->|"quorum met early"| P["Premature commit<br/>floor behind
contents"]
P -.->|"stale floor admits it"| DUP["Duplicate file<br/>same path in two
snapshots"]
classDef good fill:#dcfce7,stroke:#16a34a,color:#14532d
classDef bad fill:#fee2e2,stroke:#dc2626,color:#7f1d1d
classDef evt fill:#ffedd5,stroke:#ea580c,color:#7c2d12
class RS,F,OK good
class S,D,P,DUP bad
class R,RW,G evt
```
Three pieces:
1. `Channel.start()` registers a `ConsumerRebalanceListener` and adds a
protected `onControlPartitionsRevoked` hook (default no-op).
2. `Coordinator` overrides that hook to call a new `CommitState.reset()`.
3. `CommitState.reset()` is `clearResponses()` + `endCurrentCommit()` — it
drops both buffers, the counter, and the commit id.
State is **reset, never flushed**. Flushing control-topic offsets without
committing the files would drop data. The events stay on the control topic and
are re-read; nothing is lost.
`startTime` is deliberately preserved, so a commit interrupted mid-cycle
re-drives on the next `process()` rather than waiting out another full commit
interval.
Partitions that are *lost* rather than cleanly revoked arrive at the same
hook, since `ConsumerRebalanceListener.onPartitionsLost` delegates to
`onPartitionsRevoked` by default.
## Why this is sufficient
The invariant the sink depends on is:
> `controlTopicOffsets[P] >= max(control offset of every buffered envelope
for P) + 1`
It holds naturally within a session, because `consumeAvailable` updates the
map *before* dispatching to `receive()`. The eager rewind broke it by lowering
the map while the buffer kept the higher offsets.
`reset()` empties the buffer at the same moment the consumer rewinds, so the
replay refills both in lockstep and the recorded floor can never be written
behind its own snapshot contents again.
With `currentCommitId == null`, `addReady` takes the `!isCommitInProgress()`
branch and counts nothing, and `isCommitReady()` short-circuits to `false`.
With `commitBuffer` empty, `tableCommitMap()` has nothing to hand to
`AppendFiles`. Both paths are closed.
## The state machine, before and after
```mermaid
stateDiagram-v2
direction LR
state "BEFORE — a rebalance ends nothing" as before {
[*] --> Idle1
Idle1: Idle
Started1: Commit started
Collecting1: Collecting
Quorum1: Quorum check
Committed1: Committed
Idle1 --> Started1
Started1 --> Collecting1
Collecting1 --> Quorum1
Quorum1 --> Committed1
Rebalance1: Rebalance (unobserved)
Rebalance1 --> Collecting1: replay re-buffers
Rebalance1 --> Quorum1: id still matches
}
```
```mermaid
stateDiagram-v2
direction LR
state "AFTER — revoke is a real transition" as after {
[*] --> Idle2
Idle2: Idle
Started2: Commit started
Collecting2: Collecting
Quorum2: Quorum check
Committed2: Committed
Idle2 --> Started2
Started2 --> Collecting2
Collecting2 --> Quorum2
Quorum2 --> Committed2
Revoked2: Partitions revoked
Collecting2 --> Revoked2: rebalance
Revoked2 --> Idle2: reset(), startTime kept
}
```
The arrows invert. Before, the replay points *into* a live commit. After,
the rebalance pulls state *off* the rail and returns it to Idle.
## Tests
- `TestCommitState.testResetDiscardsInFlightState` — asserts `reset()`
clears both buffers, the counter and the commit id.
- `TestCoordinator.testControlPartitionsRevokedResetsInFlightCommit` —
drives a `MockConsumer` rebalance and asserts a fresh commit is started.
- `TestCoordinator.testControlPartitionsRevokedRewindDoesNotDoubleCount` —
models the rewind directly. Two source partitions, one reports, the partition
is revoked and reassigned, and the same `DataWritten`/`DataComplete` pair is
delivered again. Asserts no `CommitToTable` is emitted; without the reset the
stale `DataComplete` pushes readiness to 2 and a commit fires with half its
data.
Verified that both coordinator tests **fail** when `commitState.reset()` is
commented out, so they are not vacuous.
```
19 tests, 0 failures (TestCoordinator 14, TestCommitState 5)
./gradlew :iceberg-kafka-connect:spotlessCheck BUILD SUCCESSFUL
```
## Scope
Deliberately limited to the reset. Not addressed here, and better as
separate PRs:
- `controlTopicOffsets` is still written with `put()` rather than
`merge(..., Long::max)`. Making the map monotonic is independent hardening.
- `onPartitionsAssigned` is an empty listener. Seeking forward on assignment
would prevent the replay entirely, which is a different and arguably stronger
guarantee — it would also stop replayed events from distorting
`validThroughTs()`.
- The zombie-coordinator path (`CommitterImpl.hasLeaderPartition()`
returning a partial member snapshot mid-rebalance, and `startCoordinator()`
no-opping on `if (null == this.coordinatorThread)`) is untouched. #17376
targets that.
## Related
- #16282 — the report this fixes
- #13763 — same symptom, trigger never confirmed
- #13756 — coordinator lifecycle on task stop
- #15710, #15651 — per-commitId `RowDelta` separation; addresses
sequence-number collision for equality deletes, not append-only re-registration
- #17376 — coordinator election / split-brain
---
**AI Disclosure**
- Model: Claude Opus 4.6
- Platform/Tool: opencode
- Human Oversight: partially reviewed
- Prompt Summary: The source change (`Channel`, `CommitState`,
`Coordinator`, and the three tests) was written by the human author. An AI
agent was asked to analyse issue #16282 against the codebase, verify the
failure mechanism, confirm the new tests fail without the fix, and draft this
PR description and its diagrams. The description and diagrams are AI-generated
and have been partially reviewed by the author.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
