You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "Schiewatir (via GitHub)" <gi...@apache.org> on 2026/05/15 23:15:34 UTC
[PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Schiewatir opened a new pull request, #16360:
URL: https://github.com/apache/iceberg/pull/16360
## Problem
After a Kafka cluster is recreated (old cluster deleted, new one started), the Apache Iceberg Kafka Connect sink stops committing metadata. Parquet files accumulate in object storage but no snapshots are created, and the logs show `"Coordinator … found nothing to commit to table"` on every commit cycle. (Reported in #15293.)
**Root cause:** The control topic resets to offset 0 on the new cluster, but the Iceberg snapshot still stores committed offsets from the old cluster (e.g. `{"0":100}`). The deduplication filter in `Coordinator.commitToTable()` rejects every `DataWritten` event whose offset is below the stored baseline, so all data files are silently dropped.
```java
// before fix – stale minOffset = 100, envelope.offset() = 0..4 → all filtered
Long minOffset = committedOffsets.get(envelope.partition());
return minOffset == null || envelope.offset() >= minOffset;
```
## Fix
When a partition's current control-topic offset is **lower** than its stored committed offset, the committed offset for that partition is reset to the current offset minus one (so the next write is accepted). This is detected per-partition during `commitToTable()`, so there is no behaviour change for the normal (non-reset) path.
After the reset, the filter accepts the new events, a snapshot is committed, and its stored offset is reset to the new baseline — replication resumes.
Fixes #15293
--
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "MuriloBelar (via GitHub)" <gi...@apache.org>.
MuriloBelar commented on PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#issuecomment-5422952621
Still facing this issue
--
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #16360: Kafka Connect: recover from metadata commit stall after Kafka cluster recreation
URL: https://github.com/apache/iceberg/pull/16360
--
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#issuecomment-5013522702
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any 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
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "Rakshithdb (via GitHub)" <gi...@apache.org>.
Rakshithdb commented on PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#issuecomment-5328839408
@Schiewatir Looking forward for the solution.
--
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#discussion_r3360125288
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -239,6 +268,9 @@ private void commitToTable(
envelopeList.stream()
.filter(
envelope -> {
+ if (controlTopicReset) {
Review Comment:
The merged-offset computation already routes through `baseOffsets` (line 260), which is `Map.of()` when `controlTopicReset` is true. If the dedup filter below looked up `baseOffsets.get(envelope.partition())` instead of `committedOffsets`, the existing `minOffset == null` branch would already pass every envelope on reset, making this `if (controlTopicReset) return true;` removable. That folds the dedup baseline and the merge baseline into one source of truth, leaving `controlTopicReset` to gate only the warn log. The `offsetValidator` calls must keep using `committedOffsets`.
--
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#discussion_r3377402522
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -229,8 +229,37 @@ private void commitToTable(
// records for other partitions. Merge the updated topic partitions with the last committed
// offsets.
Map<Integer, Long> committedOffsets = lastCommittedOffsetsForTable(table, branch);
+
+ // Detect if the control topic was reset (e.g., after Kafka cluster recreation).
+ // If the current coordinator's observed control topic offsets are lower than the
+ // previously committed offsets stored in the snapshot, the control topic has likely
+ // been reset and the stored offsets are stale. In this case, skip deduplication
+ // to avoid silently dropping all data files and blocking metadata commits.
+ boolean controlTopicReset =
+ !committedOffsets.isEmpty()
+ && committedOffsets.entrySet().stream()
+ .anyMatch(
+ e -> {
+ Long current = controlTopicOffsets.get(e.getKey());
+ return current != null && current < e.getValue();
+ });
+
+ if (controlTopicReset) {
+ LOG.warn(
+ "Coordinator {}: detected possible Kafka cluster recreation for table {}. "
+ + "Control topic offsets {} are lower than previously committed offsets {}. "
+ + "Skipping offset deduplication and resetting stored offset baseline.",
+ taskId,
+ tableIdentifier,
+ controlTopicOffsets,
+ committedOffsets);
+ }
+
+ // When a reset is detected, base the stored offsets only on the current control topic
+ // offsets so subsequent commits use the correct (new-cluster) baseline.
+ Map<Integer, Long> baseOffsets = controlTopicReset ? Map.of() : committedOffsets;
Review Comment:
`baseOffsets = controlTopicReset ? Map.of() : committedOffsets` empties the dedup baseline for every partition as soon as any one partition looks reset, but the description states the reset is per-partition. Scoping it per partition keeps dedup where it is still valid: build `baseOffsets` by filtering `committedOffsets` to entries where `controlTopicOffsets.get(p) == null || current >= committed`, dropping the reset ones. Then both the merge and the dedup filter run off `baseOffsets` (reset partitions have no entry, so `Long::max` stores their new low offset and the `minOffset == null` branch passes their events), leaving `controlTopicReset` to gate only the warn log. This is behavior-identical for the single-partition control topic, so no regression.
--
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#issuecomment-4940758609
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: [PR] Kafka Connect: recover from metadata commit stall after Kafka cluster recreation [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#issuecomment-4666332443
Thanks for the follow-up iterations. Could you rebase onto main and drop the four Netty commits (and the three build.gradle edits), leaving this PR scoped to Coordinator.java and TestCoordinator.java? That picks up the central fix, removes the duplicate Netty pins, and keeps the diff focused.
--
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