diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java index 75e8beaef17c..8a965ca71aa7 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java @@ -265,7 +265,7 @@ public void notifyCheckpointComplete(long checkpointId) { @Override public void notifyCheckpointAborted(long checkpointId) { - if (checkpointId == this.checkpointId) { + if (checkpointId == this.checkpointId && !WriteMetadataEvent.BOOTSTRAP_INSTANT.equals(this.instant)) { executor.execute(() -> { this.ckpMetadata.abortInstant(this.instant); }, "abort instant %s", this.instant); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java index 45a4e04bab28..f059c7050ca5 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java @@ -97,8 +97,6 @@ public void close() { public void bootstrap(HoodieTableMetaClient metaClient) throws IOException { fs.delete(path, true); fs.mkdirs(path); - metaClient.getActiveTimeline().getCommitsTimeline().filterPendingExcludingCompaction() - .lastInstant().ifPresent(instant -> startInstant(instant.getTimestamp())); } public void startInstant(String instant) {