You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "ericyangliu (via GitHub)" <gi...@apache.org> on 2026/07/22 21:17:17 UTC
[I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
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: issues-unsubscribe@iceberg.apache.org.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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "vbhanuchander-lang (via GitHub)" <gi...@apache.org>.
vbhanuchander-lang commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5463577119
@ravithejatech714-create I came at this from the code rather than from an incident — I traced `Channel.consumeAvailable` on `main` and confirmed the watermark regression, but I am not operating an affected deployment, so I have no repair story to offer you. Scoping a replay range is a real problem, but it is downstream of this one; the open question on this issue is still the `put` → `max` fix in the connector.
@ericyangliu are you still planning to open the PR? You ticked that you could contribute the fix independently and said you would put one up, and I deliberately stood down because the finding is yours. Just checking it has not gone stale — nothing has touched `Channel.java` on `main` since you filed this, so the regression is still live.
Happy to leave it with you either way. If you would rather hand it off, say so and I will open it with the diagnosis credited to you.
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks closed issue #17340: Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read)
URL: https://github.com/apache/iceberg/issues/17340
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "ravithejatech714-create (via GitHub)" <gi...@apache.org>.
ravithejatech714-create commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5462939318
The connector correctness issue here belongs in Iceberg, but the resulting recovery problem is very relevant to us: after fixing the root cause, how do you identify the exact source range that needs repair without broadly replaying the topic and creating more duplicates?
ReplayGuard focuses on that scope → validation → controlled recovery workflow.
Curious how you repaired the affected Kafka range in practice.
https://replayguard.in/
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "vbhanuchander-lang (via GitHub)" <gi...@apache.org>.
vbhanuchander-lang commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5518310577
Thanks @ericyangliu — opened #17933.
It is the `put` → `merge(..., Long::max)` change you suggested, nothing more. The argument I made
for it in the PR is that `Coordinator.commitToTable` already folds `controlTopicOffsets()` with
`Long::max` against the last committed offsets, so making the map itself monotonic only brings it in
line with how its two consumers — that stamp and `commitConsumerOffsets()` — already treat it. The
forward path is unchanged.
On the open question I raised above about whether the offset fix is sufficient on its own: I still
cannot settle it from the code, so I have kept the bounded recently-committed-locations set out of
the PR and flagged it there as a separate design call. That keeps the correctness fix reviewable on
its own, and it is your call whether you want to pursue the backstop.
The diagnosis and the production evidence in this issue are yours, and the PR says so.
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "akomisarek (via GitHub)" <gi...@apache.org>.
akomisarek commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5621884545
Even with this scenario, why this piece of code doesn't protect from creating duplicates?
https://github.com/apache/iceberg/blob/2702c983668c57177eba79ec97d9103d660cdc81/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java#L265-L280
Shouldn't filtering with already committed offsets, skip the reprocessed ones? 🤔
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "vbhanuchander-lang (via GitHub)" <gi...@apache.org>.
vbhanuchander-lang commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5639629082
@akomisarek good question — that filter is narrower than it looks, in three ways.
**It is keyed on what *that table's* snapshot already recorded.** `committedOffsets` comes from `lastCommittedOffsetsForTable(table, branch)`, i.e. the `kafka.connect.offsets.<topic>.<group>` property on that table's current snapshot. So the guard only exists for a table+partition pair that has already committed at least once. The `minOffset == null` branch is a pass, not a skip: for a partition with no recorded entry — a newly added table, a table that has never seen envelopes from that control-topic partition, or the first commits after the connector is created — every envelope goes through, replays included.
**It can only be as far forward as the value that was stored.** The value written back is `max(committedOffsets, controlTopicOffsets)` per partition, so for a partition that already has an entry it cannot regress. But where there is no prior entry, whatever `controlTopicOffsets` holds at that moment is stamped in directly. That is the part #17933 was about: `consumeAvailable` recorded the consumed position with an unconditional `put(partition, offset + 1)`, so a re-read of the control topic moved the tracked position **backwards**. A regressed value then both (a) becomes the group offset a restarted channel resumes from, which is what produces the re-read in the first place, and (b) can be the value stamped onto a snapshot for a partition with no prior entry. Afterwards, replayed envelopes are compared against a watermark that is behind them and pass.
**Nothing downstream catches what slips past.** `distinctByKey(ContentFile::location)` dedupes only *within* a single commit's payload list, and `appendFiles` does no path-level comparison against what the table already references. So one envelope through the filter is one duplicate data file in the table — which is why the report shows double-referenced files rather than an error.
Worth adding that the filter also cannot see a second coordinator. If two `CoordinatorThread`s are alive for one connector (#17637), each reads the table's committed offsets at its own moment and both can pass the same envelopes. That is a separate cause with the same symptom, and the offset arithmetic here does not help with it.
So: the filter is a real guard, but it is a per-table low-water mark that only protects a partition it already knows about, and it was being fed a watermark that could move backwards. Whether your specific duplicates came through the `null` branch or a regressed watermark would show in the `kafka.connect.offsets.*` property on the snapshots around the incident — if the recorded position for the affected partition ever decreases between consecutive snapshots, that is the second case.
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "vbhanuchander-lang (via GitHub)" <gi...@apache.org>.
vbhanuchander-lang commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5287748553
I checked the mechanism against current main and the structural part of your diagnosis holds:
`Channel.consumeAvailable` records the position with a plain `put`, so the map is last-write-wins
rather than a high-water mark:
```java
// Channel.java:126
controlTopicOffsets.put(record.partition(), record.offset() + 1);
```
Nothing in that method compares against the value already stored, so any path that re-reads a
partition from an earlier position — a seek back to committed group offsets on restart, as you
describe — moves the recorded offset **backwards**. The value then flows into
`commitConsumerOffsets()` and into the `kafka.connect.offsets` stamped on the snapshot, which is what
makes the regression durable rather than transient.
Your suggested change is consistent with how the value is used everywhere else, since every consumer
of `controlTopicOffsets()` wants "furthest position consumed", never "most recently seen":
```java
controlTopicOffsets.merge(record.partition(), record.offset() + 1, Long::max);
```
Two things I could not confirm from the code alone, and which I think are the open questions for a
reviewer rather than objections:
1. Whether making the watermark monotonic is sufficient on its own, or whether the replay can also
deliver an envelope whose offset is genuinely above the last committed watermark — in which case
the min-offset filter passes it legitimately and only content-level deduplication would catch it.
Your step 5 suggests the former, but it depends on exactly where the seek lands.
2. Whether the bounded "recently committed locations" set you float is wanted at all. That is a
design call for the maintainers, and worth separating from the one-line fix so the fix is not held
up by it.
I have not opened a PR — you said you can put one up and it is your finding, so it should be yours.
Flagging the verification here in case it helps it move, since this has been open three weeks with
no response and the production evidence you gathered (149 files, ~112k duplicated rows) makes it
worth someone's attention.
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "ravithejatech714-create (via GitHub)" <gi...@apache.org>.
ravithejatech714-create commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5488283524
Thanks for clarifying — that distinction makes sense.
You're right that the watermark regression and the `put` → `max` correctness fix are the issue here, while the recovery question only becomes relevant for an operator after an affected deployment has already produced duplicates.
I appreciate the clarification, and I’ll leave this thread focused on the connector fix.
--
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: [I] Kafka Connect: duplicate data file commits after task restart (offsets watermark regresses on control topic re-read) [iceberg]
Posted by "ericyangliu (via GitHub)" <gi...@apache.org>.
ericyangliu commented on issue #17340:
URL: https://github.com/apache/iceberg/issues/17340#issuecomment-5517565021
Hi @vbhanuchander-lang , please feel free to create a PR if you'd like.
--
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