ericyangliu opened a new issue, #17340:
URL: https://github.com/apache/iceberg/issues/17340
### Apache Iceberg version
main (development)
### Query engine
Kafka Connect
### Please describe the bug 🐞
We run the Kafka Connect sink on MSK Connect (Connect runtime 3.7.0) writing
to a Glue-cataloged format-version 3 table: 320-partition source topic,
`tasks.max=160`, `iceberg.control.commit.interval-ms=60000`. Our build is a
1.11.0-SNAPSHOT from June 2026, but the relevant code in
`Channel.consumeAvailable` is unchanged on current main.
After worker rebalances that restarted tasks mid commit cycle, the sink
committed the same data files in two consecutive commits, ~60s apart. The table
ended up with two live manifest entries per affected file, so every scan reads
those files twice. We hit this six times over three days: 149 files
double-referenced, 111,927 duplicated rows, and a downstream Spark MERGE
failing on `MERGE_CARDINALITY_VIOLATION` until we repaired the manifests.
Snapshot summaries for one pair (consecutive commits, distinct commit ids):
```
commit A: added-data-files=38, kafka.connect.offsets={"0": 38584300}
commit B: added-data-files=24, kafka.connect.offsets={"0": 38584378} # 19
of the 24 identical to commit A
```
Sequence, reconstructed from worker logs and snapshot metadata:
1. Coordinator sends START_COMMIT; workers respond with DataWritten.
2. A task restart hits mid-cycle. The restarted channel seeks its control
topic consumer back to the last committed group offsets and re-reads (`Received
commit response when no commit in progress, this can happen during recovery`).
3. `Channel.consumeAvailable` tracks the consumed position with an
unconditional put:
```java
controlTopicOffsets.put(record.partition(), record.offset() + 1);
```
so the re-read regresses the map.
4. The in-flight commit lands mid re-read and stamps `kafka.connect.offsets`
with a watermark lower than the position those events had originally been
consumed at.
5. `clearResponses()` runs, but the tail of the re-read keeps arriving. The
replayed envelopes carry offsets above the regressed watermark, so the
min-offset filter in `Coordinator.commitToTable` passes them.
`distinctByKey(ContentFile::location)` only dedupes within a single commit, and
validation passes because no concurrent commit happened.
6. The next commit (a partial commit in our case, since most tasks were
still restarting) appends the replayed files again. Append does no path-level
dedup, so the table now references those files twice.
The duplicated files span many distinct writer tasks (one file per writer,
all opened within one cycle's write window), which rules out a single
misbehaving worker and points at coordinator-side replay rather than worker
re-announcement.
Suggested fix is to make the tracked position monotonic:
```java
controlTopicOffsets.merge(record.partition(), record.offset() + 1,
Long::max);
```
so replayed envelopes always fall below the recorded watermark and the
existing filter drops them, as designed. A bounded set of recently committed
file locations in the Coordinator could additionally serve as a content-level
backstop for replay cases the offset arithmetic can't see; happy to discuss
whether that's wanted.
I can put up a PR. Related recent hardening in this area: #16433, #16434.
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [x] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [ ] I cannot contribute a fix for this bug at this 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: [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]
