You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "tongwai-wong-appier (via GitHub)" <gi...@apache.org> on 2026/05/11 05:50:14 UTC
[I] [Kafka Connect] Duplicate file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
tongwai-wong-appier opened a new issue, #16282:
URL: https://github.com/apache/iceberg/issues/16282
## Apache Iceberg version
1.10.1
## Query engine
N/A (issue is in the Kafka Connect sink connector)
## Please describe the bug 🐞
### Summary
When a Kafka `Connection timed out` triggers a worker task crash and coordinator switchover, the **same parquet data file can be registered in two different Iceberg snapshots**, causing row duplication. This occurs on **append-only** tables with no equality deletes.
We have observed this in production across two independent incidents on IKC 1.10.1 with a REST catalog (backed by GCS storage).
### Mechanism
The failure sequence is:
1. Coordinator C1 initiates a commit cycle (commitId `A`).
2. Worker writes data file `X` to storage and sends `DATA_WRITTEN` to the control topic.
3. C1 receives `DATA_WRITTEN`, commits file `X` to the Iceberg table → **Snapshot S1 is created successfully**.
4. Before C1 can advance the **control-topic consumer offset** past the `DATA_WRITTEN` event, the worker hits a `Connection timed out` → task crash → coordinator shutdown.
5. A new coordinator C2 starts. Its control-topic consumer position is **before** the `DATA_WRITTEN` event for file `X` (since C1 never advanced the offset).
6. C2 re-consumes the stale `DATA_WRITTEN` for file `X`, buffers it, and commits it to the table → **Snapshot S2 is created with the same file `X`**.
7. Result: file `X` appears in both S1 and S2. Every row in that file is now read twice.
### Key observation: this is NOT fixed by PR #15710 / PR #15651
Both PRs address a related but distinct problem: stale and current `DataWritten` events mixed in a single `RowDelta`, causing sequence number collision that breaks equality-delete semantics.
In our append-only scenario, sequence numbers are irrelevant. The problem is that **Iceberg's `AppendFiles` does not check whether a `file_path` is already registered in the table**. Even with per-commitId `RowDelta` grouping, the stale event would still be committed as a separate `RowDelta`, and the file would be registered a second time.
The root issue is the **lack of idempotent file registration** in the coordinator's commit path, combined with **non-atomic offset advancement** (Iceberg commit succeeds but control-topic offset is not advanced before crash).
### Evidence
Observed on IKC 1.10.1 with a REST catalog and GCS storage, on an append-only table with no equality deletes.
#### Incident — 2026-05-07 13:00 UTC
- Duplicate groups: 4,873
- Duplicate rows: 9,746 (extra: 4,873)
- **Same physical file registered in two snapshots:**
```
Snapshot 8264179290764999750 @ 2026-05-07T13:14:36 UTC
Snapshot 314603727896153559 @ 2026-05-07T13:36:44 UTC
```
- The file was written by the worker at ~13:14 (confirmed via timestamp embedded in the file name).
- Between 13:15 and 13:35, the log shows **20 consecutive minutes of `Commit timeout reached`** — two coordinators racing for the optimistic lock on the metadata file:
```
13:12:40 Cannot commit: metadata location ...532037... has changed from ...532038...
13:12:47 Cannot commit: metadata location ...532038... has changed from ...532039...
...
13:14:22 WorkerSinkTask ... Error: Connection timed out
13:14:22 Committer lost leader partition. Stopping Coordinator.
13:14:33 Cannot commit: metadata location ...532046... has changed from ...532047...
13:14:42 Commit a52ae0ca complete, snapshot 8264179290764999750, valid-through 2026-05-07T13:14:02.856Z
← Snapshot S1 succeeds (old coordinator's commit)
13:15:33 – 13:34:34 Commit timeout reached (×20, one per minute)
← New coordinator cannot commit because old coordinator's zombie is still racing
13:35:02 New coordinator discovers group coordinator, resets offsets
13:36:44 Snapshot 314603727896153559 committed
← Snapshot S2: same file re-registered
```
### Confirmed via metadata inspection
Using the `data_files` metadata table with time travel:
```sql
WITH base AS (
SELECT *,
COUNT(1) OVER (PARTITION BY file_path) AS _file_count
FROM <table>.data_files
TIMESTAMP AS OF '<snapshot_timestamp>'
)
SELECT file_path, record_count, _file_count
FROM base
WHERE _file_count >= 2
```
Result confirms: **identical `file_path`, identical `record_count`, appearing in two different snapshots**.
### Possible fix directions (our speculation — happy to discuss)
From our analysis, the fundamental issue is that file registration via `AppendFiles` / `RowDelta` is **not idempotent** — appending a file that is already in the table creates a duplicate manifest entry rather than a no-op. Combined with the non-atomic relationship between Iceberg commit and control-topic offset advancement, any coordinator crash between these two steps will cause re-registration on recovery.
We are not sure which approach would be most appropriate within the IKC codebase, but some directions we considered:
1. **Commit-level file dedup**: Before committing, check if any of the pending `file_path`s are already registered in the current table snapshot. Skip files that are already present.
2. **Stale event filtering by committed offset**: On coordinator startup, read the table's last committed offsets and discard any `DataWritten` events whose offsets fall within the already-committed range.
We'd appreciate guidance from maintainers on the preferred approach.
### Related issues
- #13763 — Same symptom (duplicate file path in Iceberg metadata snapshot). The exact trigger was never confirmed in that issue. Our investigation provides a concrete trigger path (coordinator switchover due to Kafka timeout) and evidence that PR #15710 / #15651 do not address the append-only case.
- #13756 — Merged to `main` (targeting 1.11, not backported to 1.10.x). Fixes coordinator lifecycle (force-close on task stop) but does not prevent the duplicate commit itself.
- #15710, #15651 — Both open, addressing per-commitId `RowDelta` separation. Fixes sequence number collision for equality-delete tables but does **not** prevent file re-registration in append-only mode (see explanation above).
### Willingness to contribute
We are willing to contribute a fix or test case if the maintainers can confirm the preferred approach.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "tongwai-wong-appier (via GitHub)" <gi...@apache.org>.
tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5087157214
We took a different fix on 1.11.0. Two parts.
**Rewind** — a `ConsumerRebalanceListener` that seeks forward instead of letting the replay happen:
```java
@Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
for (TopicPartition partition : partitions) {
Long inMemoryOffset = controlTopicOffsets.get(partition.partition());
if (inMemoryOffset != null && inMemoryOffset > consumer.position(partition)) {
consumer.seek(partition, inMemoryOffset); // + a warn
}
}
}
```
This does strictly more than `merge(..., Long::max)`. That keeps the floor from regressing, but the
replay still happens — the replayed `DataComplete` still double-count in `readyBuffer`, so the commit
still fires early. No duplicate files, but `validThroughTs()` then takes `min(timestamp)` over only
the partitions that did report, and the ones that didn't aren't in the list to make
`allMatch(timestamp != null)` false, so the snapshot claims a `valid-through-ts` later than the true
watermark. Seeking forward means no replay at all, so buffer / offsets map / ready count stay
consistent. Skipping past the group's committed offset is safe because those records were consumed
by this same `Channel` — their envelopes are in `commitBuffer` and get committed from there.
**Zombie** — we kept the existing election but made *de-election* deterministic: record the leader
partition at election time and have `close()` check `closedPartitions.contains(leaderPartition)`,
instead of re-running `hasLeaderPartition()`. That re-query is the thing that returns an empty or
partial member snapshot mid-rebalance and leaves the coordinator running. Much smaller than your
rework and it removes the same root cause; your `save()`-level reconcile is the stronger version
since it also self-heals missed callbacks.
Two notes on the PR:
- `put()` → `merge(..., Long::max)` is an independent fix for #16282 — worth saying so in the
description so it isn't dropped as unrelated to the election rework.
- `isCommitReady()` still compares `receivedPartitionCount`, which counts duplicates. You fixed the
right-hand side; that's the left half of the same quorum. `validThroughTs()` also doesn't filter
`readyBuffer` by `commitId` while `isCommitReady()` does.
Branch with both plus regression tests:
https://github.com/emlynazuma/iceberg/tree/tongwai/ikc-1.11.0-rebalance-fixes
Haven't run your PR against our repro setup yet — will report back when I do.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5090461421
Thanks @tongwai-wong-appier
This split-brain become a bigger problem if operating in the multi-parition control topic (a point which I raised during the discussion with bryan as having multi-partition control topic does not help but having single partition control topic actually solves many of the problems). I have simplified the PR for now so that it does not faces much scrutiny for merge. The larger refactoring I am planning to take as a separate PR. I am testing the changes but it would help if you can test those and let me know if any other change or corner case if I have missed.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "tongwai-wong-appier (via GitHub)" <gi...@apache.org>.
tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5082896086
Here's the timeline you asked for. No `seek()` is involved — all it takes is a member entering or
leaving `<group>-coord`, and a zombie coordinator is exactly such a member.
Two rebalances to keep apart:
- **source side** — the sink tasks' consumer group over the source topic partitions
- **coord side** — the coordinator's own consumer group `<group>-coord` over the control topic
They're coupled: the coordinator runs on whichever task owns the lowest source partition
(`containsFirstPartition` → `findFirstTopicPartition`), so every source-side rebalance that moves
that partition starts or stops a coordinator — i.e. changes coord-side membership. And source-side
rebalances flap.
Take d1 → d2 → d1, a single flap of the lowest source partition:
- **d1 loses it.** `close(closedPartitions)` → `hasLeaderPartition(closedPartitions)` describes the
group to find the first partition; during a rebalance that snapshot can come back empty or
partial, so it returns false and `stopCoordinator()` is skipped. **Coordinator C keeps running.**
- **d2 gains it.** `open()` → `startCoordinator()` → **C' is constructed** → its first `poll()` sends
JoinGroup → coord-side rebalance.
- **d1 gains it back.** `open()` → `startCoordinator()`, but `if (null == this.coordinatorThread)`
is false, so it's a **no-op**. C carries straight on, with its `commitBuffer`, `currentCommitId`
and `controlTopicOffsets` intact.
- **d2 loses it.** `close()` → `stopCoordinator()` → `consumer.close()` → LeaveGroup → coord-side
rebalance again.
Across the whole flap C is never reconstructed — the no-op guard keeps it alive — while C' joins and
leaves around it. IKC never sets `partition.assignment.strategy`, so the client default
`[RangeAssignor, CooperativeStickyAssignor]` takes its protocol from the first entry: **eager**.
Eager revokes every partition and discards its fetch position, even for a member that gets the same
partition straight back; `updateFetchPositions()` then refetches the committed offset. That's the
rewind, and it happens inside the consumer — which is why there's no `.seek()` to find.
Nothing on that path touches C's in-memory state, and no `ConsumerRebalanceListener` is registered.
That's t2: `commitBuffer` 0–7, `controlTopicOffsets` `{0:8}`, position back at 0. The replayed
records carry the current `commitId` (the group offset only advances on a successful `doCommit()`),
so `isCommitReady()` fires mid-drain at 8+2 = 10 and the commit runs while the map is still `{0:2}`.
Your co-reset invariant holds for C' — the Coordinator that was constructed. C is the one that
breaks, and it was never constructed.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "tongwai-wong-appier (via GitHub)" <gi...@apache.org>.
tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5099171550
Makes sense on single- vs multi-partition control topic — glad seek-forward and the recorded
`leaderTopicPartition` check in `close()` both made it into the simplified version; that covers the
rewind and the zombie path we walked through above.
Still haven't run it against our repro — will do and report back.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5083420717
Yes this:
> d1 loses it. close(closedPartitions) → hasLeaderPartition(closedPartitions) describes the
group to find the first partition; during a rebalance that snapshot can come back empty or
partial, so it returns false and stopCoordinator() is skipped. Coordinator C keeps running.
d2 gains it. open() → startCoordinator() → C' is constructed → its first poll() sends
JoinGroup → coord-side rebalance.
d1 gains it back. open() → startCoordinator(), but if (null == this.coordinatorThread)
is false, so it's a no-op. C carries straight on, with its commitBuffer, currentCommitId
and controlTopicOffsets intact.
d2 loses it. close() → stopCoordinator() → consumer.close() → LeaveGroup → coord-side
rebalance again.
is an issue of split-brain I flagged much earlier and is also targeted in the PR(https://github.com/apache/iceberg/pull/17376). This PR ensures that no state change happens between Kafka Connect calls open/close and the stuff that is performed in open and close. The current descertibe members and earlier state check was the actual bug and removal of that only partially fixed. All this can be caused by the having multiple coordinator and having a multipartition control topic makes this situation adverse (a point which I raised in discussion with brain). Also I am going to remove the task id from the transaction id of the coordinator which will fence the other coordinator once a new one is elected. All of this will make the situation more intact.
Also thanks for pointing I think this can be a corner case which will be solved by the PR changes as it will enforce a single coordinator as we will not longer check the minPartition by describing members but by just taking the partition "0" of the smallest (lexicographically) topic from the subscription .
Let me know if any more guard you think is required.
Also I would request you to if possible take the changes from this PR and test these. Meanwhile I will also test these changes against the production load.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5077253784
I think this is not possible `C2 re-consumes the stale DATA_WRITTEN for file X, buffers it, and commits it to the table → Snapshot S2 is created with the same file X.` as any coordinator before doing any metadata commit checks in the iceberg the last control topic offset. If the records consumed from the control topic by the coordinator is behind what is already committed then that record is dropped and will not land in the next snapshot.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "nahidupa (via GitHub)" <gi...@apache.org>.
nahidupa commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5522487917
Hi @tongwai-wong-appier, @kumarpritam863
We are also observing similar behavior. I’ve put together the following PR to address the issue. Please have a look and share your feedback or comments.
[Kafka Connect: Reset in-flight commit state on control-topic rebalance](https://github.com/apache/iceberg/pull/17925)
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "tongwai-wong-appier (via GitHub)" <gi...@apache.org>.
tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5077459104
Thanks @kumarpritam863 — you're right, and the mechanism I wrote in the issue body is wrong. The
floor check has been there since #10351, so it's in 1.10.1 too, and it does drop the stale
`DATA_WRITTEN` in the case I described.
But could the recorded offset itself go stale? `controlTopicOffsets` is overwritten rather than
max-merged, so it moves *backwards* whenever the coordinator's control-topic consumer rewinds:
```java
// Channel.java
controlTopicOffsets.put(record.partition(), record.offset() + 1); // plain put()
```
and that map — not the envelopes being committed — is what the snapshot's floor is derived from:
```java
// Coordinator.java
Map<Integer, Long> mergedOffsets =
Stream.of(committedOffsets, controlTopicOffsets)
.flatMap(map -> map.entrySet().stream())
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Long::max));
```
Single control-topic partition, commit expects 10 partitions, table floor starts at 1. Offsets are
as stored by the code (`record.offset() + 1`):
| | position | `controlTopicOffsets` | in buffer | ready count | result |
|---|---|---|---|---|---|
| t1 | consumed 1–8 | `{0:9}` | 1–8 | 8 | not ready |
| t2 | **rewinds to 1** | `{0:9}` | 1–8 | 8 | — |
| t3 | re-reads 1, 2 | **`{0:3}`** | 1–8 | 8+2 = **10** → commit | floor is 1, so envelopes 1–8 are appended, but the summary records `max(1, 3) = 3` |
| t4 | reads 3–10 | `{0:11}` | 3–10 | — | — |
| t5 | next commit | | 3–20 | | floor is 3, so 3–20 all pass → files from 3–8 appended again |
The snapshot at t3 holds files up to control offset 8 but advertises a floor of 3. Once the
recorded floor lags the snapshot's own contents, the filter can't drop those events on the next
cycle — and `commitConsumerOffsets()` reads the same map, so the group offset is pushed backwards
too, keeping the window open.
`validateWith(offsetValidator(...))` from #14510 doesn't seem to close this: at t5 both
`expectedOffsets` and `lastCommittedOffsets(baseSnapshots)` are `3`, so validation passes. It
guards against a concurrent writer during retry, not against a floor that was already written
behind its own snapshot.
On how the rewind gets triggered:
- It needs neither two coordinators nor the early commit at t3 — `isCommitTimedOut() ->
commit(true)` in the same window has the same effect.
- The `-coord` consumer runs with `enable.auto.commit=false`, so any rejoin resets its position to
the last `commitConsumerOffsets()`.
Does this hold, or am I missing a guard?
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5099961901
@tongwai-wong-appier
Agreed, but what I was thinking is shouldn't this change sufficient along with open/close and zombie coordinator gurad changes that is there in the PR:
```
protected void consumeAvailable(Duration pollDuration) {
ConsumerRecords<String, byte[]> records = consumer.poll(pollDuration);
while (!records.isEmpty()) {
for (ConsumerRecord<String, byte[]> record : records) {
// the consumer stores the offset of the next record to consume,
// so increment the record offset by one
if (record.offset() < controlTopicOffsets.getOrDefault(record.partition(), 0L)) {
LOG.warn("Channel {} processed an already processed offset {} for partition {}", taskId, record.offset(), record.partition());
continue;
}
controlTopicOffsets.put(record.partition(), record.offset() + 1);
Event event = AvroUtil.decode(record.value());
if (event.groupId().equals(connectGroupId)) {
LOG.debug("Received event of type: {}", event.type().name());
if (receive(new Envelope(event, record.partition(), record.offset()))) {
LOG.info("Handled event of type: {}", event.type().name());
}
}
}
records = consumer.poll(pollDuration);
}
}
```
As neither the worker not the coordinator would want to read a record which is already read by them in any of the scenarios.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5099965669
I m making this change and will test to see if anything is there which can not be covered by this.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5081883420
The committed offsets come from the Iceberg latest created snapshot:
` private Map<Integer, Long> lastCommittedOffsetsForTable(Table table, String branch) {
Snapshot snapshot = latestSnapshot(table, branch);
if (snapshot == null) {
return Map.of();
}
Iterable<Snapshot> branchAncestry =
SnapshotUtil.ancestorsOf(snapshot.snapshotId(), table::snapshot);
return lastCommittedOffsets(branchAncestry);
}`
Even if the controlTopic goes backward, it will be ignored as it will be less than what is stored in the table.
Any control topic only contributes in any of the snapshot only if it's offset is strictly greater than what already committed in the last created snapshot.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "tongwai-wong-appier (via GitHub)" <gi...@apache.org>.
tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5082228154
The point isn't whether events below the table offset get dropped — it's that **the table offset
itself is wrong**:
- the snapshot written at t3 records `{"0": 2}` but contains files through offset 7, so it no longer
describes its own contents
- a pre-existing floor doesn't save it either — the rewind lands *on* it, so `Long::max` adopts the
regressed value
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5082450519
Yeah I got your concern but I do not see how that can happen as:
1. No seek, no intra-session rewind. There is no .seek() anywhere on the control-topic consumer (grep confirms: the only seek is KafkaUtils.seekToLastCommittedOffsets on the source
consumer in CommitterImpl.java:120, unrelated to the floor). consumeAvailable only ever calls consumer.poll(), whose offsets are monotonically increasing per
partition within a session. So within one Channel, controlTopicOffsets[P] is monotonically non-decreasing — the plain put is effectively a max because records arrive in order.
2. The only way to resume at a lower control offset is a new Coordinator, and that resets both structures together:
- startCoordinator → new Coordinator(...) → new CommitState(config) (Coordinator.java:114) ⇒ fresh empty commitBuffer.
- new Coordinator → super(...) → new Channel ⇒ fresh empty controlTopicOffsets (Channel.java:54, a final field never cleared during a Channel's life).
So the buffer and the offsets map are never from different sessions. At commit time the invariant
▎ controlTopicOffsets[P] ≥ max(control offset of every buffered envelope for P) + 1
always holds, and Long::max with committedOffsets only raises the floor further. The recorded floor is never below the snapshot's own contents. Your t3 state (buffer = 0–7,
controlTopicOffsets = {0:2}) has no code path that produces it.
Walking your timeline
- t1 {0:8}, buffer 0–7 — reachable, normal.
- t2 "rewinds to 0", buffer still 0–7 — impossible. A rewind means a new Coordinator/Channel, which gives you controlTopicOffsets = {} and an empty buffer, not {0:8}/buffer-0–7 with a
rewound consumer. Consumption and buffering can't desynchronize.
- Everything from t3 onward inherits the impossible t2 state.
You're right that the validator wouldn't catch it
Your read of offsetValidator (from #14510) is correct — at t5 both expectedOffsets and lastCommittedOffsets would be 2, so it passes. It guards against a concurrent writer during retry,
not a self-inflicted low floor. It's simply not the thing preventing this — the monotonic-consumption + co-reset invariant above is.
Also to handle the split-brain scenario which can cause multiple coordinators or even no coordinator in some cases, I am opening a PR where I will mention this.
Also can you properly give the timeline where the backward shift is possible which I may be missing. If that is a real issue I would be more than happy to include that in the same PR.
--
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 file registration across snapshots during coordinator recovery (append-only, IKC 1.10.1) [iceberg]
Posted by "tongwai-wong-appier (via GitHub)" <gi...@apache.org>.
tongwai-wong-appier commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5082844093
Assume a zombie coordinator exists — the split-brain case your PR is about, where a previous
leader's `close()` never reached `stopCoordinator()`, so two coordinators are members of
`<group>-coord` and the healthy one holds the control-topic partition.
The rewind doesn't need a `seek()` — it needs a *different* member to leave the group.
- Zombie shuts down: `CommitterImpl.close()` → `stopCoordinator()` → `CoordinatorThread.terminate()`
→ `Channel.stop()` → `consumer.close()` → **LeaveGroup**
- Membership change → `<group>-coord` rebalances → the *surviving* coordinator has to rejoin, even
though it's healthy and keeps the same partition
- No `partition.assignment.strategy` set → client default `[RangeAssignor, CooperativeStickyAssignor]`
→ protocol comes from the first entry = **eager** → all partitions revoked, fetch positions
discarded → `updateFetchPositions()` refetches the committed offset. That's the rewind, inside the
consumer — nothing to grep for.
- Nothing is constructed on that path and no `ConsumerRebalanceListener` is registered, so
`commitBuffer` / `currentCommitId` / `controlTopicOffsets` are untouched. **That's t2**: buffer 0–7,
`{0:8}`, position 0.
- Replayed records carry the current `commitId` (the group offset only advances on a successful
`doCommit()`), so `isCommitReady()` fires mid-drain at 8+2 = 10 — the commit runs while the map is
`{0:2}`.
Your co-reset invariant covers the Coordinator being constructed, not the one already running.
--
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