You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "LouisDeconinck (via GitHub)" <gi...@apache.org> on 2026/09/14 07:43:43 UTC
[PR] Flink: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
LouisDeconinck opened a new pull request, #18097:
URL: https://github.com/apache/iceberg/pull/18097
Part of #16445
All APIs tagged `@deprecated ... will be removed in 1.12.0` have already been removed on main. This PR removes the remaining code path documented for removal in 1.12: `DynamicCommitter`'s per-checkpoint `List<CommitRequest>` handling.
`DynamicCommitter` grouped multiple `CommitRequest`s per (table, branch, checkpoint) triplet only to process Flink state from releases before 1.11, where the upstream `DynamicWriteResultAggregator` emitted multiple committables per checkpoint. The code comment scheduled this for removal in 1.12, since upgrading through 1.11 migrates the state to a single commit request per checkpoint.
Changes:
- `commit()` now tracks a single `CommitRequest<DynamicCommittable>` per checkpoint (matching `IcebergCommitter`) and fails fast via `Preconditions.checkState` if a duplicate is received, instead of silently overwriting it and later signaling it as committed without its data files being committed.
- `commitPendingRequests` iterates a single request per checkpoint directly.
- Tests `testTableBranchAtomicCommitForAppendOnlyData` and `testCommitDeltaTxnWithAppendFiles` now pack the multiple delta manifests into one `DynamicCommittable`, preserving the same commit coverage.
- Applied identically to Flink 1.20, 2.1, 2.2 and 2.3.
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "LouisDeconinck (via GitHub)" <gi...@apache.org>.
LouisDeconinck commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044749410
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -222,24 +220,22 @@ private static long getMaxCommittedCheckpointId(
private void commitPendingRequests(
Table table,
String branch,
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> commitRequestMap,
+ NavigableMap<Long, CommitRequest<DynamicCommittable>> commitRequestMap,
String newFlinkJobId,
String operatorId)
throws IOException {
long checkpointId = commitRequestMap.lastKey();
List<ManifestFile> manifests = Lists.newArrayList();
NavigableMap<Long, List<WriteResult>> pendingResults = Maps.newTreeMap();
Review Comment:
Done in dede42c — pending results are now NavigableMap<Long, WriteResult>, built per checkpoint via WriteResult.Builder; replacePartitions/commitDeltaTxn consume the merged results directly. Applied identically to the v1.20, v2.1, v2.2, and v2.3 implementations.
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #18097: Flink: Remove multi-commit-request state handling in DynamicCommitter
URL: https://github.com/apache/iceberg/pull/18097
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044190527
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -222,24 +220,22 @@ private static long getMaxCommittedCheckpointId(
private void commitPendingRequests(
Table table,
String branch,
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> commitRequestMap,
+ NavigableMap<Long, CommitRequest<DynamicCommittable>> commitRequestMap,
String newFlinkJobId,
String operatorId)
throws IOException {
long checkpointId = commitRequestMap.lastKey();
List<ManifestFile> manifests = Lists.newArrayList();
NavigableMap<Long, List<WriteResult>> pendingResults = Maps.newTreeMap();
Review Comment:
This should be simplified to `NavigableMap<Long, WriteResult>` and every method down the call stack should accept a single `WriteResult` per checkpoint.
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044825549
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -125,26 +125,26 @@ public void commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
Each (table, branch, checkpoint) triplet must have only one commit request.
There may be commit requests from previous checkpoints which have not been committed yet.
-
- We currently keep a List of commit requests per checkpoint instead of a single CommitRequest<DynamicCommittable>
- to process the Flink state from previous releases, which had multiple commit requests created by the upstream
- DynamicWriteResultAggregator. Iceberg 1.12 will remove this, and users should upgrade to the 1.11 release first
- to migrate their state to a single commit request per checkpoint.
*/
- Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>>
+ Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, CommitRequest<DynamicCommittable>>>>
commitRequestMap = Maps.newHashMap();
for (CommitRequest<DynamicCommittable> request : commitRequests) {
DynamicCommittable committable = request.getCommittable();
- commitRequestMap
- .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
- .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
- .computeIfAbsent(committable.checkpointId(), unused -> Lists.newArrayList())
- .add(request);
+ CommitRequest<DynamicCommittable> previous =
+ commitRequestMap
+ .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
+ .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
+ .put(committable.checkpointId(), request);
+ Preconditions.checkState(
+ previous == null,
+ "Received multiple commit requests for table %s branch %s at checkpoint %s",
Review Comment:
Jobs upgrading directly from 1.10 or earlier versions would hit this precondition and fail. If we are going down this path, let's make the error message more descriptive with actionable instructions for users on how to resolve this:
- Option 1: users must upgrade to 1.11 first. TLDR: The 1.11 committer commits or skips the old-format committables during initializeState and removes them from the collector, so any savepoint taken from the 1.11 job is already clean. One successful restore plus one savepoint is enough. Then upgrade to 1.12.
- Option 2 (unsure if this option would work across all Flink versions given FLINK-37605): stop-with-savepoint + `--drain` flag and start job with no state.
More generally, I think the cleanup is good but it comes with the cost of potentially creating some annoyance/loss of state for end users. @pvary what's your opinion?
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044825549
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -125,26 +125,26 @@ public void commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
Each (table, branch, checkpoint) triplet must have only one commit request.
There may be commit requests from previous checkpoints which have not been committed yet.
-
- We currently keep a List of commit requests per checkpoint instead of a single CommitRequest<DynamicCommittable>
- to process the Flink state from previous releases, which had multiple commit requests created by the upstream
- DynamicWriteResultAggregator. Iceberg 1.12 will remove this, and users should upgrade to the 1.11 release first
- to migrate their state to a single commit request per checkpoint.
*/
- Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>>
+ Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, CommitRequest<DynamicCommittable>>>>
commitRequestMap = Maps.newHashMap();
for (CommitRequest<DynamicCommittable> request : commitRequests) {
DynamicCommittable committable = request.getCommittable();
- commitRequestMap
- .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
- .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
- .computeIfAbsent(committable.checkpointId(), unused -> Lists.newArrayList())
- .add(request);
+ CommitRequest<DynamicCommittable> previous =
+ commitRequestMap
+ .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
+ .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
+ .put(committable.checkpointId(), request);
+ Preconditions.checkState(
+ previous == null,
+ "Received multiple commit requests for table %s branch %s at checkpoint %s",
Review Comment:
Jobs upgrading directly from 1.10 or earlier versions would hit this precondition and fail. If we are going down this path, let's make the error message more descriptive with actionable instructions for users on how to resolve this:
- Option 1: users must upgrade to 1.11 first. TLDR: The 1.11 committer commits or skips the old-format committables during initializeState and removes them from the collector, so any savepoint taken from the 1.11 job is already clean. One successful restore plus one savepoint is enough. Then upgrade to 1.12.
- Option 2 (unsure if this option would work across all Flink versions given FLINK-37605): stop-with-savepoint + `--drain` flag and start job with no state.
More generally, I think the cleanup is good but it comes with the cost of potentially creating some annoyance/loss of state for end users so I can also lean towards keeping this committable merging code for longer. @pvary what's your opinion?
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#issuecomment-6457214011
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044825549
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -125,26 +125,26 @@ public void commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
Each (table, branch, checkpoint) triplet must have only one commit request.
There may be commit requests from previous checkpoints which have not been committed yet.
-
- We currently keep a List of commit requests per checkpoint instead of a single CommitRequest<DynamicCommittable>
- to process the Flink state from previous releases, which had multiple commit requests created by the upstream
- DynamicWriteResultAggregator. Iceberg 1.12 will remove this, and users should upgrade to the 1.11 release first
- to migrate their state to a single commit request per checkpoint.
*/
- Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>>
+ Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, CommitRequest<DynamicCommittable>>>>
commitRequestMap = Maps.newHashMap();
for (CommitRequest<DynamicCommittable> request : commitRequests) {
DynamicCommittable committable = request.getCommittable();
- commitRequestMap
- .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
- .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
- .computeIfAbsent(committable.checkpointId(), unused -> Lists.newArrayList())
- .add(request);
+ CommitRequest<DynamicCommittable> previous =
+ commitRequestMap
+ .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
+ .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
+ .put(committable.checkpointId(), request);
+ Preconditions.checkState(
+ previous == null,
+ "Received multiple commit requests for table %s branch %s at checkpoint %s",
Review Comment:
Jobs upgrading directly from 1.10 or earlier versions would hit this precondition and fail. If we are going down this path, let's make the error message more descriptive with actionable instructions for users on how to resolve this:
- Option 1: users must upgrade to 1.11 first. TLDR: The 1.11 committer commits or skips the old-format committables during initializeState and removes them from the collector, so any savepoint taken from the 1.11 job is already clean. One successful restore plus one savepoint is enough. Then upgrade to 1.12.
- Option 2 (unsure if this option would work across all Flink versions given FLINK-37605): stop-with-savepoint + `--drain` flag and start job with no state.
We would also have to document these upgrade requirements somewhere, maybe in release notes of 1.12?
More generally, I think the cleanup is good but it comes with the cost of potentially creating some annoyance/loss of state for end users so I can also lean towards keeping this committable merging code for longer (given it's not much extra code). @pvary what's your opinion?
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044190527
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -222,24 +220,22 @@ private static long getMaxCommittedCheckpointId(
private void commitPendingRequests(
Table table,
String branch,
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> commitRequestMap,
+ NavigableMap<Long, CommitRequest<DynamicCommittable>> commitRequestMap,
String newFlinkJobId,
String operatorId)
throws IOException {
long checkpointId = commitRequestMap.lastKey();
List<ManifestFile> manifests = Lists.newArrayList();
NavigableMap<Long, List<WriteResult>> pendingResults = Maps.newTreeMap();
Review Comment:
This should be simplified to `NavigableMap<Long, WriteResult>` and the loop below should use the `WriteResult.Builder` to merge multiple `WriteResult`s from each `FlinkManifestUtil.readCompletedFiles()` call. Every method down the call stack should accept this simplified collection: `replacePartitions`, `commitDeltaTxn`.
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#issuecomment-5680763349
@aiborodin: Could you please review?
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044190527
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -222,24 +220,22 @@ private static long getMaxCommittedCheckpointId(
private void commitPendingRequests(
Table table,
String branch,
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> commitRequestMap,
+ NavigableMap<Long, CommitRequest<DynamicCommittable>> commitRequestMap,
String newFlinkJobId,
String operatorId)
throws IOException {
long checkpointId = commitRequestMap.lastKey();
List<ManifestFile> manifests = Lists.newArrayList();
NavigableMap<Long, List<WriteResult>> pendingResults = Maps.newTreeMap();
Review Comment:
This should be simplified to `NavigableMap<Long, WriteResult>` and every method down the call stack should accept `NavigableMap<Long, WriteResult> pendingResults`. The loop below should use the `WriteResult.Builder` to merge multiple `WriteResult`s from each `FlinkManifestUtil.readCompletedFiles()` call.
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#issuecomment-5727081611
@aiborodin: Given that 1.12 is basically out, we can cheaply keep it a bit longer and remove it only to 1.13 release.
The code is much nicer without this, so I would remove as soon as we released 1.12.0
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044190527
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -222,24 +220,22 @@ private static long getMaxCommittedCheckpointId(
private void commitPendingRequests(
Table table,
String branch,
- NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> commitRequestMap,
+ NavigableMap<Long, CommitRequest<DynamicCommittable>> commitRequestMap,
String newFlinkJobId,
String operatorId)
throws IOException {
long checkpointId = commitRequestMap.lastKey();
List<ManifestFile> manifests = Lists.newArrayList();
NavigableMap<Long, List<WriteResult>> pendingResults = Maps.newTreeMap();
Review Comment:
This should be simplified to `NavigableMap<Long, WriteResult>` and every method down the call stack should accept this collection: `replacePartitions`, `commitDeltaTxn`. The loop below should use the `WriteResult.Builder` to merge multiple `WriteResult`s from each `FlinkManifestUtil.readCompletedFiles()` call.
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#discussion_r4044825549
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -125,26 +125,26 @@ public void commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
Each (table, branch, checkpoint) triplet must have only one commit request.
There may be commit requests from previous checkpoints which have not been committed yet.
-
- We currently keep a List of commit requests per checkpoint instead of a single CommitRequest<DynamicCommittable>
- to process the Flink state from previous releases, which had multiple commit requests created by the upstream
- DynamicWriteResultAggregator. Iceberg 1.12 will remove this, and users should upgrade to the 1.11 release first
- to migrate their state to a single commit request per checkpoint.
*/
- Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>>
+ Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, CommitRequest<DynamicCommittable>>>>
commitRequestMap = Maps.newHashMap();
for (CommitRequest<DynamicCommittable> request : commitRequests) {
DynamicCommittable committable = request.getCommittable();
- commitRequestMap
- .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
- .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
- .computeIfAbsent(committable.checkpointId(), unused -> Lists.newArrayList())
- .add(request);
+ CommitRequest<DynamicCommittable> previous =
+ commitRequestMap
+ .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
+ .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
+ .put(committable.checkpointId(), request);
+ Preconditions.checkState(
+ previous == null,
+ "Received multiple commit requests for table %s branch %s at checkpoint %s",
Review Comment:
Jobs upgrading directly from 1.10 or earlier versions would hit this precondition and fail. If we are going down this path, let's make the error message more descriptive with actionable instructions for users on how to resolve this:
- Option 1: users must upgrade to 1.11 first. TLDR: The 1.11 committer commits or skips the old-format committables during initializeState and removes them from the collector, so any savepoint taken from the 1.11 job is already clean. One successful restore plus one savepoint is enough. Then upgrade to 1.12.
- Option 2 (unsure if this option would work across all Flink versions given FLINK-37605): stop-with-savepoint + `--drain` flag and start job with no state.
We would also have to document these upgrade requirements somewhere, maybe in release notes of 1.12?
More generally, I think the cleanup is good but it comes with the cost of potentially creating some annoyance/loss of state for end users so I can also lean towards keeping this committable merging code for longer. @pvary what's your opinion?
--
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: Remove multi-commit-request state handling in DynamicCommitter [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18097:
URL: https://github.com/apache/iceberg/pull/18097#issuecomment-6280378330
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