You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "UrsSchoenenbergerNu (via GitHub)" <gi...@apache.org> on 2026/04/08 14:29:49 UTC
[PR] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
UrsSchoenenbergerNu opened a new pull request, #15913:
URL: https://github.com/apache/iceberg/pull/15913
Closes #15846
### Quick Summary
When running with unaligned checkpoints, data files may reach IcebergFilesCommitter after the associated checkpoint's first barrier.
We then need to account for these files in the state for the 'next' checkpoint to avoid discarding them during recovery or failed checkpoints.
### Details
We have recently encountered a case of data loss in an application using FlinkSink (v1), running on Flink 2.x with unaligned checkpoints.
What we saw:
One commit to the Iceberg table was missing approximately half of the data files that were written by the writer during this checkpoint.
Around this time, one checkpoint was started and Flink snapshots were triggered, but this checkpoint timed out before completing.
Based on our research, we believe that the culprit might be the internal state tracking in IcebergFilesCommitter. Here's our current theory:
For unaligned checkpoints, the order of operations happening on IcebergFilesCommitter can change subtly since processElement can be called for a FlinkWriteResult that is part of checkpoint N even after snapshotState was called for checkpoint N:
processElement(FlinkWriteResult1[part of checkpoint N]), recorded in writeResultsSinceLastSnapshot
snapshotState(checkpoint N)
writeToManifestUptoLatestCheckpoint(N) is called, putting the contents of writeResultsSinceLastSnapshot to dataFilesPerCheckpoint(N) and clearing writeResultsSinceLastSnapshot
processElement(FlinkWriteResult2[part of checkpoint N]), recorded in writeResultsSinceLastSnapshot (now the single element there)
checkpoint N times out, so notifyCheckpointComplete(N) is never called
processElement(FlinkWriteResult3[part of checkpoint N+1])
snapshotState(checkpoint N+1)
writeToManifestUptoLatestCheckpoint(N+1) is called, putting the contents of writeResultsSinceLastSnapshot to dataFilesPerCheckpoint(N) and dataFilesPerCheckpoint(N+1) and clearing writeResultsSinceLastSnapshot. dataFilesPerCheckpoint(N) now loses FlinkWriteResult1 even though it was never committed and contains only FlinkWriteResult2
notifyCheckpointComplete(checkpoint N+1)
Iceberg commit is started, but contains only FlinkWriteResult2 and FlinkWriteResult3, not FlinkWriteResult1.
This sequence, or equivalently one where checkpoint N does not time out, but completes after checkpoint N+1, therefore leads to data loss.
This explanation aligns with the effects that we're seeing. The root cause seems to be that snapshotState->writeToManifestUpToLatestCheckpoint->writeResultsSinceLastSnapshot seems to implicitly assume that all records for the checkpoints have already been processed when snapshotState() is called - i.e. it assumes aligned checkpoints. If this assumption breaks, AND in addition a later checkpoint is snapshotted before an earlier one was notified complete, the issue described above is observed.
Additionally, we suspect there's a second failure mode with unaligned checkpoints that loses data on job recovery: If our above theory is correct, then IcebergFilesCommitter has an issue with elements for checkpoint N being processed after snapshotState(N). But the iceberg commit triggered during notifyComplete(N) only commits the records from the snapshot. On recovery, this means one of two things: Either initializeState()->.tailMap(N) loses information about not-yet-committed records; or even if it remembered to do so, it would think that these records should be committed as part of Flink checkpoint N. It feels like there's a problem when a checkpoint ID can appear in dataFilesPerCheckpoint after it has already been used as maxCommittedCheckpointId, and the strict tailMap exclusion has no way to distinguish "already committed" from "deferred post-barrier" data.
### Contents of this PR
This PR contains three test cases that trigger the test harness in ways that are only possible when using unaligned checkpoints. Without the associated code change in prod code, these test cases fail.
### Disclaimer / disclosure as per contributing guidelines
AI tools were used to initially pinpoint the possible issue and draft test cases. The code was subsequently rewritten condensed and clarified by hand. To the best of my knowledge, I believe that the way in which the reproduction test cases trigger the harness are all cases that DO happen with unaligned checkpoints, and the data loss that we had encountered at runtime aligns with the one that happens in these reproduction test cases.
### Caveats
The logic that I'm changing here was last changed by https://github.com/apache/iceberg/pull/10526 while fixing a data duplication bug. The test cases that were added for that bug are still green with the changes that I'm proposing here. I would still like to tag @pvary and @zhongqishang and kindly ask them for advice on this issue here and this associated PR, as I'm sure they have a much deeper understanding of the inner workings of `IcebergFileCommitter` than 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: [PR] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#issuecomment-4464769146
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3066168737
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
Review Comment:
NIT
```suggestion
* In case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
+ return sourceCheckpointIdAlreadyCommitted || sourceCheckpointIdHasDataInSnapshot
+ ? checkpointId
+ : sourceCheckpointId;
Review Comment:
Is this correct under all scenarios? This will mean that we include files of an older snapshot into the current checkpoint, but IMHO they need to be in the following snapshot (`sourceCheckpointId + 1`), not in the latest. There can be outstanding WriteResults for multiple snapshots, which have not been committed to Iceberg.
Otherwise, we violate the order in case of deletions.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
Review Comment:
```suggestion
// ID will be chosen.
```
NIT
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "UrsSchoenenbergerNu (via GitHub)" <gi...@apache.org>.
UrsSchoenenbergerNu commented on PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#issuecomment-5144277480
@pvary It seems to me there is no operator-level config or annotation where we could mark IcebergSink as not supporting unaligned checkpoints, so the only option might be to introspect the checkpointing config in open() and throwing when someone tries to run the sink unaligned? This would be a breaking change for existing users though - potentially a preferable break, but there might be users who have never experienced this data loss. Do you think there might be a better safeguard here? Documentation only is likely to be read only after someone stumbles over lost data.
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "UrsSchoenenbergerNu (via GitHub)" <gi...@apache.org>.
UrsSchoenenbergerNu commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3079409101
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
+ return sourceCheckpointIdAlreadyCommitted || sourceCheckpointIdHasDataInSnapshot
+ ? checkpointId
+ : sourceCheckpointId;
Review Comment:
Hm, that's a great question and I'm not sure I understand all the intricacies of it.
Let me clarify please if I understand your point correctly: by just using `checkpointId` we might run into a case where for the next commit, the deletes might have been pre-barrier and the data post-barrier or the other way round, creating an inconsistent commit?
That seems like a variation of the problem that #10526 tried to fix. The tests introduced in that PR pass with my modifications, so the scenario must be slightly more complicated.
Regarding your proposed fix of `sourceCheckpointId + 1` works, I am worried that this might produce new data loss victims due to the `tailMap` call on operator recovery. At least in my head, I had convinced myself that only `checkpointId` (current barrier) is safe enough to not lose any data on recovery.
On a higher level, I am wondering whether there even is an implementation that can maintain consistent commits when faced with both equality delete files and unaligned checkpoints. Isn't there something inherently at odds with these two concepts as we can always have deletions pre-barrier and data post-barrier or the other way round?
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3066078886
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
Review Comment:
When post-barrier `WriteResult`s include equality deletes (V2 tables), they get merged with the next checkpoint's data and committed in a single `RowDelta`. This changes the sequence number of those deletes compared to aligned mode. The behavior should still be correct (deletes end up with a higher sequence and still apply to earlier data), but it would be good to have an explicit test for this scenario, e.g. `testPostBarrierEqualityDeleteFiles` with format v2.
Also: since `commitDeltaTxn` commits each checkpoint's `RowDelta` separately (#10526), post-barrier equality deletes that get redirected here end up in the same `RowDelta` as the current checkpoint's data+deletes. Worth confirming this doesn't cause duplication issues similar to the ones #10526 fixed.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
Review Comment:
Nit: Javadoc should start with a capital letter.
```suggestion
* In case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
Review Comment:
Nit: remove the blank line after the opening brace (inconsistent with the rest of the file).
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
+ return sourceCheckpointIdAlreadyCommitted || sourceCheckpointIdHasDataInSnapshot
Review Comment:
Nit: reflow the comment to avoid the orphaned `// ID` line:
```suggestion
// For aligned checkpoints, both conditions will be false and the upstream operator's
// checkpoint ID will be chosen.
```
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java:
##########
@@ -1076,6 +1076,220 @@ public void testSpecEvolution() throws Exception {
}
}
+ /**
+ * (unaligned checkpoints) With unaligned checkpoints a writer subtask may deliver its write
+ * result after {@code snapshotState(N)} has already fired. ("post-barrier data"). This test
+ * verifies that post-barrier data for a checkpoint that never completes is not lost, it must be
+ * committed together with the next successful checkpoint. Also covers the case where the
+ * successful checkpoint itself has post-barrier data, which must then wait for the checkpoint
+ * after that.
+ *
+ * <pre>
+ * processElement(dataA, checkpointId=1)
+ * snapshotState(1)
+ * processElement(dataB, checkpointId=1)
+ * // checkpoint 1 never completes
+ * processElement(dataC, checkpointId=2)
+ * snapshotState(2)
+ * processElement(dataD, checkpointId=2)
+ * notifyCheckpointComplete(2) // commits dataA, dataB, dataC
+ * snapshotState(3)
+ * notifyCheckpointComplete(3) // commits dataD
+ * </pre>
+ */
+ @TestTemplate
+ public void testPostBarrierDataSurvivesFailedCheckpoint() throws Exception {
+ long timestamp = 0;
+ JobID jobId = new JobID();
+ OperatorID operatorId;
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
+ harness.setup();
+ harness.open();
+ operatorId = harness.getOperator().getOperatorID();
+
+ assertSnapshotSize(0);
+ assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
+
+ RowData rowA = SimpleDataUtil.createRowData(1, "early-checkpoint1");
+ DataFile dataFileA = writeDataFile("data-A", ImmutableList.of(rowA));
+ RowData rowB = SimpleDataUtil.createRowData(2, "post-barrier-checkpoint1");
+ DataFile dataFileB = writeDataFile("data-B", ImmutableList.of(rowB));
+ RowData rowC = SimpleDataUtil.createRowData(3, "early-checkpoint2");
+ DataFile dataFileC = writeDataFile("data-C", ImmutableList.of(rowC));
+ RowData rowD = SimpleDataUtil.createRowData(4, "post-barrier-checkpoint2");
+ DataFile dataFileD = writeDataFile("data-D", ImmutableList.of(rowD));
+
+ long checkpoint1 = 1;
+ long checkpoint2 = 2;
+ long checkpoint3 = 3;
+
+ harness.processElement(of(checkpoint1, dataFileA), ++timestamp);
+ harness.snapshot(checkpoint1, ++timestamp);
+ assertFlinkManifests(1);
+
+ // post-barrier, arrives after snapshotState(1); checkpoint1 then FAILS (no notify)
+ harness.processElement(of(checkpoint1, dataFileB), ++timestamp);
+
+ harness.processElement(of(checkpoint2, dataFileC), ++timestamp);
+ harness.snapshot(checkpoint2, ++timestamp);
Review Comment:
Nit: existing tests consistently verify manifest counts after `snapshot()` and `notifyOfCompletedCheckpoint()`. Consider adding `assertFlinkManifests(2)` here (one manifest from checkpoint 1's snapshot, one new manifest for checkpoint 2) for consistency.
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#issuecomment-4253757008
We have discussed unaligned checkpoints with @gyfora previously and if I remember correctly we decided that the unaligned checkpoint concept does not really allow committing a consistent state to the Iceberg table. We might be able to work around this by adding some extra checks/barriers but the end result will be something like an aligned checkpoint implemented by hand, so we decided that it doesn't seem to worth the effort.
If we all agree on this, we might update the documentation to explicitly state that aligned checkpoints are a requirement for a Flink Sink. Maybe even add a check which would prevent adding a Flink Sink when unaligned checkpoint is used?
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "UrsSchoenenbergerNu (via GitHub)" <gi...@apache.org>.
UrsSchoenenbergerNu commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3079409101
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
+ return sourceCheckpointIdAlreadyCommitted || sourceCheckpointIdHasDataInSnapshot
+ ? checkpointId
+ : sourceCheckpointId;
Review Comment:
Let me clarify please if I understand your point correctly: by just using `checkpointId` we might run into a case where for the next commit, the deletes might have been pre-barrier and the data post-barrier or the other way round, creating an inconsistent commit?
That seems like a variation of the problem that #10526 tried to fix. The tests introduced in that PR pass with my modifications, so the scenario must be slightly more complicated.
Regarding your proposed fix of `sourceCheckpointId + 1` works, I am worried that this might produce new data loss victims due to the `tailMap` call on operator recovery. At least in my head, I had convinced myself that only `checkpointId` (current barrier) is safe enough to not lose any data on recovery.
On a higher level, I am wondering whether there even is an implementation that can maintain consistent commits when faced with both equality delete files and unaligned checkpoints. Isn't there something inherently at odds with these two concepts as we can always have deletions pre-barrier and data post-barrier or the other way round?
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "vbhanuchander-lang (via GitHub)" <gi...@apache.org>.
vbhanuchander-lang commented on PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#issuecomment-5527925621
@UrsSchoenenbergerNu this was closed by the stale bot on 23 May, not on merit, and #15846 is still
open. @fmorillo7694 independently reproduced it at AWS three days ago. I verified your fix against
current `main` and found one gap that matters for the 1.12 cut — posting both so a committer has
less to redo.
**Verified against `main` (`0107f19a`), first-hand:**
- the patch applies cleanly, no conflicts
- with it, `TestIcebergFilesCommitter` is **114 tests, 0 failures**
- reverting *only* `IcebergFilesCommitter.java` and keeping your tests gives **18 failures**, in
exactly `testPostBarrierDataSurvivesFailedCheckpoint`,
`testPostBarrierDataForFailedCheckpointSurvivesRecovery` and
`testPostBarrierDataMergedWithEarlyDataOnRecovery`
18 is the same count @fmorillo7694 reported, which is a good sign the two reproductions are the same
thing rather than two different ones.
**The gap: the patch only covers `flink/v2.1`.** `main` now has **v1.20, v2.1, v2.2 and v2.3**, and
the vulnerable class is in all four — all five references to `writeResultsSinceLastSnapshot` are
present in each. `IcebergFilesCommitter.java` is **byte-identical across v2.1, v2.2 and v2.3**
(same sha), and differs only in v1.20. I copied your patched file to v2.2 and v2.3 unchanged and
`:iceberg-flink:iceberg-flink-2.3:compileJava` is green, so for those two it is a verbatim port;
v1.20 needs the change adapted by hand.
That matters because @fmorillo7694's point is that **`flink/v2.3` is what would ship the bug in
1.12** — and as it stands this PR would fix v2.1 and leave v2.3 broken. The tests would want porting
across as well, since they currently exist only under v2.1.
**@mxm** — your question from 10 April, whether the V2 `IcebergSink` is affected, was answered by
@UrsSchoenenbergerNu on 14 April: `IcebergWriteAggregator` takes the checkpoint id from a different
place, so V2 looks unaffected. Would you be willing to reopen this? Given a data-loss bug with two
independent reproductions and a release cut approaching, it seems worth not restarting from scratch.
@UrsSchoenenbergerNu it is your fix and your analysis, and it should stay yours. If you would rather
not pick it back up, I am happy to do the mechanical part — port it across the four version dirs and
the tests with it, under your authorship — just say so. And if I have misread the version matrix,
please correct me.
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "UrsSchoenenbergerNu (via GitHub)" <gi...@apache.org>.
UrsSchoenenbergerNu commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3079840893
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
Review Comment:
fixed
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#issuecomment-4523571441
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #15913: Flink: FlinkSink data loss with unaligned checkpoints
URL: https://github.com/apache/iceberg/pull/15913
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3066078886
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
Review Comment:
When post-barrier `WriteResult`s include equality deletes (V2 tables), they get merged with the next checkpoint's data and committed in a single `RowDelta`. This changes the sequence number of those deletes compared to aligned mode. The behavior should still be correct (deletes end up with a higher sequence and still apply to earlier data), but it would be good to have an explicit test for this scenario, e.g. `testPostBarrierEqualityDeleteFiles` with format v2.
Also: since `commitDeltaTxn` commits each checkpoint's `RowDelta` separately (#10526), post-barrier equality deletes that get redirected here end up in the same `RowDelta` as the current checkpoint's data+deletes. Worth confirming this doesn't cause duplication issues similar to the ones #10526 fixed.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
Review Comment:
Nit: remove the blank line after the opening brace (inconsistent with the rest of the file).
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
+ return sourceCheckpointIdAlreadyCommitted || sourceCheckpointIdHasDataInSnapshot
Review Comment:
Nit: reflow the comment to avoid the orphaned `// ID` line:
```suggestion
// For aligned checkpoints, both conditions will be false and the upstream operator's
// checkpoint ID will be chosen.
```
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java:
##########
@@ -1076,6 +1076,220 @@ public void testSpecEvolution() throws Exception {
}
}
+ /**
+ * (unaligned checkpoints) With unaligned checkpoints a writer subtask may deliver its write
+ * result after {@code snapshotState(N)} has already fired. ("post-barrier data"). This test
+ * verifies that post-barrier data for a checkpoint that never completes is not lost, it must be
+ * committed together with the next successful checkpoint. Also covers the case where the
+ * successful checkpoint itself has post-barrier data, which must then wait for the checkpoint
+ * after that.
+ *
+ * <pre>
+ * processElement(dataA, checkpointId=1)
+ * snapshotState(1)
+ * processElement(dataB, checkpointId=1)
+ * // checkpoint 1 never completes
+ * processElement(dataC, checkpointId=2)
+ * snapshotState(2)
+ * processElement(dataD, checkpointId=2)
+ * notifyCheckpointComplete(2) // commits dataA, dataB, dataC
+ * snapshotState(3)
+ * notifyCheckpointComplete(3) // commits dataD
+ * </pre>
+ */
+ @TestTemplate
+ public void testPostBarrierDataSurvivesFailedCheckpoint() throws Exception {
+ long timestamp = 0;
+ JobID jobId = new JobID();
+ OperatorID operatorId;
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
+ harness.setup();
+ harness.open();
+ operatorId = harness.getOperator().getOperatorID();
+
+ assertSnapshotSize(0);
+ assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
+
+ RowData rowA = SimpleDataUtil.createRowData(1, "early-checkpoint1");
+ DataFile dataFileA = writeDataFile("data-A", ImmutableList.of(rowA));
+ RowData rowB = SimpleDataUtil.createRowData(2, "post-barrier-checkpoint1");
+ DataFile dataFileB = writeDataFile("data-B", ImmutableList.of(rowB));
+ RowData rowC = SimpleDataUtil.createRowData(3, "early-checkpoint2");
+ DataFile dataFileC = writeDataFile("data-C", ImmutableList.of(rowC));
+ RowData rowD = SimpleDataUtil.createRowData(4, "post-barrier-checkpoint2");
+ DataFile dataFileD = writeDataFile("data-D", ImmutableList.of(rowD));
+
+ long checkpoint1 = 1;
+ long checkpoint2 = 2;
+ long checkpoint3 = 3;
+
+ harness.processElement(of(checkpoint1, dataFileA), ++timestamp);
+ harness.snapshot(checkpoint1, ++timestamp);
+ assertFlinkManifests(1);
+
+ // post-barrier, arrives after snapshotState(1); checkpoint1 then FAILS (no notify)
+ harness.processElement(of(checkpoint1, dataFileB), ++timestamp);
+
+ harness.processElement(of(checkpoint2, dataFileC), ++timestamp);
+ harness.snapshot(checkpoint2, ++timestamp);
Review Comment:
Nit: existing tests consistently verify manifest counts after `snapshot()` and `notifyOfCompletedCheckpoint()`. Consider adding `assertFlinkManifests(2)` here (one manifest from checkpoint 1's snapshot, one new manifest for checkpoint 2) for consistency.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
Review Comment:
Nit: Javadoc should start with a capital letter.
```suggestion
* In case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
```
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "UrsSchoenenbergerNu (via GitHub)" <gi...@apache.org>.
UrsSchoenenbergerNu commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3079839086
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
Review Comment:
fixed
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#issuecomment-5167332957
I think it would be good to talk this through with other users like @swapna267, @Guosmilesmile and @mxm
--
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] Flink: FlinkSink data loss with unaligned checkpoints [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #15913:
URL: https://github.com/apache/iceberg/pull/15913#discussion_r3087683873
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -423,21 +423,53 @@ public void endInput() throws IOException {
}
private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws IOException {
- if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+
+ if (!dataFilesPerCheckpoint.containsKey(checkpointId)) {
dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
}
- for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
- writeResultsSinceLastSnapshot.entrySet()) {
- dataFilesPerCheckpoint.put(
- writeResultsOfCheckpoint.getKey(),
- writeToManifest(writeResultsOfCheckpoint.getKey(), writeResultsOfCheckpoint.getValue()));
+ Map<Long, List<WriteResult>> pendingWriteResults = Maps.newHashMap();
+ for (Map.Entry<Long, List<WriteResult>> entry : writeResultsSinceLastSnapshot.entrySet()) {
+ long assignedCheckpointId = computeCheckpointId(checkpointId, entry);
+ pendingWriteResults
+ .computeIfAbsent(assignedCheckpointId, k -> Lists.newArrayList())
+ .addAll(entry.getValue());
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> entry : pendingWriteResults.entrySet()) {
+ dataFilesPerCheckpoint.put(entry.getKey(), writeToManifest(entry.getKey(), entry.getValue()));
}
// Clear the local buffer for current checkpoint.
writeResultsSinceLastSnapshot.clear();
}
+ /**
+ * in case of unaligned checkpoints, data files that were part of checkpoint N in the writer may
+ * have to become part of a later checkpoint in the committer if:
+ *
+ * <ul>
+ * <li>previous files were already committed for checkpoint N. We have to keep the manifests for
+ * new files under a later key, otherwise they are discarded during recovery after a crash
+ * <li>we already have a manifest of files to be committed for checkpoint N, even though it
+ * might not have been committed yet. In this case, we must not overwrite the manifests we
+ * already have, and we must keep them consistent with our checkpoint
+ * </ul>
+ */
+ private long computeCheckpointId(long checkpointId, Map.Entry<Long, List<WriteResult>> entry) {
+ long sourceCheckpointId = entry.getKey();
+
+ boolean sourceCheckpointIdAlreadyCommitted = sourceCheckpointId <= maxCommittedCheckpointId;
+ boolean sourceCheckpointIdHasDataInSnapshot =
+ dataFilesPerCheckpoint.containsKey(sourceCheckpointId);
+ // for aligned checkpoints, both conditions will be false and the upstream operator's checkpoint
+ // ID
+ // will be chosen.
+ return sourceCheckpointIdAlreadyCommitted || sourceCheckpointIdHasDataInSnapshot
+ ? checkpointId
+ : sourceCheckpointId;
Review Comment:
I think you're right. To be honest, I don't have a ton of experience with unaligned checkpoints, but it looks like it violates the fundamental assumption that we can flush any processed records before the checkpoint completes. If we cannot guarantee that, deletes can be applied too late, e.g. in the next snapshot, which is not correct. We can only guarantee this if we wait on some kind of barrier, which is exactly what the regular synchronous checkpointing does.
--
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