mxm commented on code in PR #16011:
URL: https://github.com/apache/iceberg/pull/16011#discussion_r3116711181
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java:
##########
@@ -128,8 +159,8 @@ public void prepareSnapshotPreBarrier(long checkpointId)
throws IOException {
new DynamicCommittable(
entries.getKey(),
writeToManifests(entries.getKey().tableName(),
entries.getValue(), checkpointId),
- getContainingTask().getEnvironment().getJobID().toString(),
- getRuntimeContext().getOperatorUniqueID(),
+ flinkJobId,
+ operatorId,
checkpointId),
Review Comment:
This code is part of `prepareSnapshotPreBarrier`, which is triggered before
the snapshot barrier passes this operator. At this point in time, the job id is
still valid and will be persisted in DynamicCommittable. Upon restore, we will
use the old JobID in DynamicCommitter to check and commit, which is correct.
--
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]
