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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
