This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new e84d229691 Flink: Do not drop IcebergSink commits after a stateless
restart (#18123)
e84d229691 is described below
commit e84d229691020ed6587df641da47e04621b1ab1a
Author: Beutlin <[email protected]>
AuthorDate: Thu Oct 8 17:32:42 2026 +0200
Flink: Do not drop IcebergSink commits after a stateless restart (#18123)
---
.../iceberg/flink/sink/IcebergCommitter.java | 25 +++++++--
.../org/apache/iceberg/flink/sink/IcebergSink.java | 1 +
.../org/apache/iceberg/flink/sink/SinkUtil.java | 2 +-
.../iceberg/flink/sink/TestIcebergCommitter.java | 64 ++++++++++++++++++++++
4 files changed, 86 insertions(+), 6 deletions(-)
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergCommitter.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergCommitter.java
index 878ddd8f13..f681fff2bf 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergCommitter.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergCommitter.java
@@ -79,6 +79,7 @@ class IcebergCommitter implements
Committer<IcebergCommittable> {
private ExecutorService workerPool;
private int continuousEmptyCheckpoints = 0;
private final boolean tableMaintenanceEnabled;
+ private final boolean isRestored;
private final int subtaskId;
IcebergCommitter(
@@ -90,6 +91,7 @@ class IcebergCommitter implements
Committer<IcebergCommittable> {
String sinkId,
IcebergFilesCommitterMetrics committerMetrics,
boolean tableMaintenanceEnabled,
+ boolean isRestored,
int subtaskId) {
this.branch = branch;
this.snapshotProperties = snapshotProperties;
@@ -97,6 +99,7 @@ class IcebergCommitter implements
Committer<IcebergCommittable> {
this.committerMetrics = committerMetrics;
this.tableLoader = tableLoader;
this.tableMaintenanceEnabled = tableMaintenanceEnabled;
+ this.isRestored = isRestored;
this.subtaskId = subtaskId;
// IcebergSink#addPreCommitTopology routes all committables to subtask 0
via a .global()
@@ -138,13 +141,25 @@ class IcebergCommitter implements
Committer<IcebergCommittable> {
}
IcebergCommittable last =
commitRequestMap.lastEntry().getValue().getCommittable();
+ // A stateless start has no committed checkpoints; the table's mark would
drop every commit.
long maxCommittedCheckpointId =
- SinkUtil.getMaxCommittedCheckpointId(table, last.jobId(),
last.operatorId(), branch);
+ isRestored
+ ? SinkUtil.getMaxCommittedCheckpointId(table, last.jobId(),
last.operatorId(), branch)
+ : SinkUtil.INITIAL_CHECKPOINT_ID;
// Mark the already committed FilesCommittable(s) as finished
- commitRequestMap
- .headMap(maxCommittedCheckpointId, true)
- .values()
- .forEach(CommitRequest::signalAlreadyCommitted);
+ NavigableMap<Long, CommitRequest<IcebergCommittable>> alreadyCommitted =
+ commitRequestMap.headMap(maxCommittedCheckpointId, true);
+ if (!alreadyCommitted.isEmpty()) {
+ LOG.warn(
+ "Skipping {} already committed checkpoint(s) up to {} for table {},
branch {}: {}",
+ alreadyCommitted.size(),
+ maxCommittedCheckpointId,
+ table.name(),
+ branch,
+ alreadyCommitted.keySet());
+ }
+
+ alreadyCommitted.values().forEach(CommitRequest::signalAlreadyCommitted);
NavigableMap<Long, CommitRequest<IcebergCommittable>> uncommitted =
commitRequestMap.tailMap(maxCommittedCheckpointId, false);
if (!uncommitted.isEmpty()) {
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
index f208ddd620..789bb00082 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
@@ -250,6 +250,7 @@ public class IcebergSink
sinkId,
metrics,
maintenanceEnabled,
+ context.getRestoredCheckpointId().isPresent(),
context.getTaskInfo().getIndexOfThisSubtask());
}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java
index d4c3d3beb8..994fac65f6 100644
--- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java
+++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java
@@ -45,7 +45,7 @@ import org.slf4j.LoggerFactory;
@Internal
public class SinkUtil {
- private static final long INITIAL_CHECKPOINT_ID = -1L;
+ static final long INITIAL_CHECKPOINT_ID = -1L;
public static final String FLINK_JOB_ID = "flink.job-id";
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
index 965ed01708..a079b0ca2a 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
@@ -238,6 +238,56 @@ class TestIcebergCommitter extends TestBase {
}
}
+ @TestTemplate
+ public void testCommitTxnAfterStatelessRestart() throws Exception {
+ RowData rowFromPreviousRun = SimpleDataUtil.createRowData(0, "hello0");
+ DataFile dataFileFromPreviousRun =
+ writeDataFile("data-previous-run",
ImmutableList.of(rowFromPreviousRun));
+
+ try (OneInputStreamOperatorTestHarness<
+ CommittableMessage<IcebergCommittable>,
CommittableMessage<IcebergCommittable>>
+ preRestartHarness = getTestHarness()) {
+ preRestartHarness.open();
+ processElement(jobId, 5, preRestartHarness, 1, OPERATOR_ID,
dataFileFromPreviousRun);
+ preRestartHarness.notifyOfCompletedCheckpoint(5);
+ }
+
+ assertSnapshotSize(1);
+ assertMaxCommittedCheckpointId(jobId, 5);
+
+ List<RowData> rows = Lists.newArrayList(rowFromPreviousRun);
+ // A stateless restart opens a fresh operator instance without restoring
prior state, so
+ // IcebergSink#createCommitter must see an empty
context.getRestoredCheckpointId().
+ try (OneInputStreamOperatorTestHarness<
+ CommittableMessage<IcebergCommittable>,
CommittableMessage<IcebergCommittable>>
+ afterRestartHarness = getTestHarness()) {
+ afterRestartHarness.open();
+
+ RowData rowAfterRestart = SimpleDataUtil.createRowData(1, "hello1");
+ DataFile dataFileAfterRestart =
+ writeDataFile("data-after-restart",
ImmutableList.of(rowAfterRestart));
+ processElement(jobId, 1, afterRestartHarness, 1, OPERATOR_ID,
dataFileAfterRestart);
+ afterRestartHarness.notifyOfCompletedCheckpoint(1);
+ rows.add(rowAfterRestart);
+ assertSnapshotSize(2);
+ assertMaxCommittedCheckpointId(jobId, 1);
+ SimpleDataUtil.assertTableRows(table, ImmutableList.copyOf(rows),
branch);
+ }
+ }
+
+ @TestTemplate
+ public void testCommitTxnSkipsAlreadyCommittedCheckpointAfterRestore()
throws Exception {
+ RowData rowFromPreviousRun = SimpleDataUtil.createRowData(1, "hello1");
+ commitCheckpoint(getCommitter(), 5, "data-previous-run",
rowFromPreviousRun);
+ assertMaxCommittedCheckpointId(jobId, 5);
+
+ commitCheckpoint(getCommitter(), 5, "data-replayed",
SimpleDataUtil.createRowData(2, "hello2"));
+
+ assertSnapshotSize(1);
+ assertMaxCommittedCheckpointId(jobId, 5);
+ SimpleDataUtil.assertTableRows(table,
ImmutableList.of(rowFromPreviousRun), branch);
+ }
+
@TestTemplate
public void testOrderedEventsBetweenCheckpoints() throws Exception {
// It's possible that two checkpoints happen in the following orders:
@@ -1347,6 +1397,10 @@ class TestIcebergCommitter extends TestBase {
// ------------------------------- Utility Methods
--------------------------------
private IcebergCommitter getCommitter() {
+ return getCommitter(true);
+ }
+
+ private IcebergCommitter getCommitter(boolean isRestored) {
IcebergFilesCommitterMetrics metric =
mock(IcebergFilesCommitterMetrics.class);
return new IcebergCommitter(
tableLoader,
@@ -1357,9 +1411,19 @@ class TestIcebergCommitter extends TestBase {
"sinkId",
metric,
false,
+ isRestored,
0);
}
+ private void commitCheckpoint(
+ IcebergCommitter committer, long checkpointId, String filename, RowData
row)
+ throws IOException, InterruptedException {
+ WriteResult writeResult = of(writeDataFile(filename,
ImmutableList.of(row)));
+ committer.commit(
+ Lists.newArrayList(
+ buildCommitRequestFor(jobId, checkpointId,
Lists.newArrayList(writeResult))));
+ }
+
private Committer.CommitRequest<IcebergCommittable> buildCommitRequestFor(
String myJobID, long checkpoint, Collection<WriteResult> writeResults)
throws IOException {
IcebergCommittable commit =
