gong commented on code in PR #8367:
URL: https://github.com/apache/inlong/pull/8367#discussion_r1246363541
##########
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`
+ rollbackAndRecover(restoredFlinkJobId, restoredCheckpointId);
+ }
+ }
+
+ private void rollback(long snapshotId) {
+ table.manageSnapshots().rollbackTo(snapshotId).commit();
+ }
+
+ private void recover(String restoredFlinkJobId, NavigableMap<Long, byte[]>
uncommittedManifests) throws Exception {
+ if (!uncommittedManifests.isEmpty()) {
+ // Committed all uncommitted data files from the old flink job to
iceberg table.
+ long maxUncommittedCheckpointId = uncommittedManifests.lastKey();
+ commitUpToCheckpoint(uncommittedManifests, restoredFlinkJobId,
maxUncommittedCheckpointId);
}
}
+ private void rollbackAndRecover(String restoredFlinkJobId, Long
restoredCheckpointId) throws Exception {
+ // 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);
+ // Find snapshot associated with restoredCheckpointId
+ long snapshotId = getSnapshotIdAssociatedWithChkId(table,
restoredFlinkJobId, restoredCheckpointId);
+
+ // Once maxCommitted CheckpointId is greater than
restoredCheckpointId, it means more data added, it need
+ // rollback
+ if (restoredCheckpointId < maxCommittedCheckpointId) {
+ if (snapshotId != -1) {
+ LOG.info("Rollback committed snapshot to {}", snapshotId);
+ rollback(snapshotId); // TODO:what if rollback throw Exception
+ } else {
+ long minUncommittedCheckpointId =
dataFilesPerCheckpoint.keySet().stream().min(Long::compareTo).get();
+ if (maxCommittedCheckpointId >= minUncommittedCheckpointId) {
+ LOG.warn("It maybe has some repeat data between chk[{},
{}]", minUncommittedCheckpointId, maxCommittedCheckpointId);
+ }
+
+ // should recover all manifest that has not been deleted. Not
deleted mean it may not be committed.
+ long uncommittedChkId =
findLastCommittedManifest(dataFilesPerCheckpoint.headMap(maxCommittedCheckpointId,
true), table.io());
+ LOG.info("Snapshot has been expired. Recover all uncommitted
snapshot between chk[{}, {}]. "
+ + "maxCommittedCheckpointId is {},
minUncommittedCheckpointId is {}.",
+ uncommittedChkId, restoredCheckpointId,
+ maxCommittedCheckpointId,
minUncommittedCheckpointId);
+ if (uncommittedChkId != -1) {
Review Comment:
-1 change to INITIAL_CHECKPOINT_ID
--
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]
