aiborodin commented on code in PR #14810:
URL: https://github.com/apache/iceberg/pull/14810#discussion_r2663979684


##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -126,9 +125,14 @@ public void 
commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
       return;
     }
 
-    // For every table and every checkpoint, we store the list of 
to-be-committed
-    // DynamicCommittable.
-    // There may be DynamicCommittable from previous checkpoints which have 
not been committed yet.
+    /*
+      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 due to a bug in the upstream
+      DynamicWriteResultAggregator. We should replace this with a single 
commit request in the next major release.

Review Comment:
   We can do this. The only caveat is that I don't know how to programmatically 
enforce users to upgrade with `stop-with-savepoint`. If we simplify the code 
and remove lists in this change, the committer state would still be compatible 
and have the same operator ID. So if users upgrade without a prior 
`stop-with-savepoint`, the committer would skip some committables if there were 
more than one in the state per (table, branch, checkpoint).  So we need some 
way to prevent accidental upgrades without a prior `stop-with-savepoint`.
   
   An alternative, potentially simpler solution would be to release this change 
in another patch release: 1.10.2. It would allow us to have a version that 
automatically migrates the state. We could then remove lists as a separate 
change in 1.11.0 and add logic to detect the old state with multiple 
committabels per (table, branch, checkpoint), which would fail the job and ask 
users to upgrade to the previous version first (1.10.2) to migrate the state.
   What do you think?



-- 
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]

Reply via email to