thexiay commented on code in PR #8367:
URL: https://github.com/apache/inlong/pull/8367#discussion_r1250290560
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java:
##########
@@ -170,27 +175,79 @@ public void initializeState(FunctionInitializationContext
context) throws Except
this.checkpointsState =
context.getOperatorStateStore().getListState(stateDescriptor);
this.jobIdState =
context.getOperatorStateStore().getListState(jobIdDescriptor);
// New table doesn't have state, so it doesn't need to do restore
operation.
- if (context.isRestored() && jobIdState.get().iterator().hasNext()) {
+ if (context.isRestored()) {
+ if (!jobIdState.get().iterator().hasNext()) {
+ LOG.error("JobId is null, Skip restore process");
+ return;
+ }
String restoredFlinkJobId = jobIdState.get().iterator().next();
+
this.dataFilesPerCheckpoint.putAll(checkpointsState.get().iterator().next());
+ // every datafiles will be added into state, so there must be data
and nullpoint exception will not happen
+ Long restoredCheckpointId =
dataFilesPerCheckpoint.keySet().stream().max(Long::compareTo).get();
Preconditions.checkState(!Strings.isNullOrEmpty(restoredFlinkJobId),
"Flink job id parsed from checkpoint snapshot shouldn't be
null or empty");
- // Since flink's checkpoint id will start from the
max-committed-checkpoint-id + 1 in the new flink job even
- // if it's restored from a snapshot created by another different
flink job, so it's safe to assign the max
- // committed checkpoint id from restored flink job to the current
flink job.
- this.maxCommittedCheckpointId = getMaxCommittedCheckpointId(table,
restoredFlinkJobId);
-
- NavigableMap<Long, byte[]> uncommittedDataFiles = Maps
- .newTreeMap(checkpointsState.get().iterator().next())
- .tailMap(maxCommittedCheckpointId, false);
- if (!uncommittedDataFiles.isEmpty()) {
- // Committed all uncommitted data files from the old flink job
to iceberg table.
- long maxUncommittedCheckpointId =
uncommittedDataFiles.lastKey();
- commitUpToCheckpoint(uncommittedDataFiles, restoredFlinkJobId,
maxUncommittedCheckpointId);
- }
+ // ------------------------------
+ // ↓ ↑
+ // a --> a+1 --> a+2 --> ... --> a+n
+ // max checkpoint id = m
+ // a >= m: supplementary commit snapshot between checkpoint (`m`,
`a`]
+ // a < m: rollback to snapshot associated with checkpoint `a`
Review Comment:
ok, it may be can be config as optional later.
--
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]
