lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4311] Fix Flink lose data on some rollback scene (#5950)

Browse Source
This commit is contained in:
吴祥平
2022-06-27 16:09:44 +08:00
committed by GitHub
parent 26c967bac6
commit 3a1fd22841
2 changed files with 1 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
View File

@@ -271,7 +271,7 @@ public class StreamWriteOperatorCoordinator
@Override @Override
public void notifyCheckpointAborted(long checkpointId) { public void notifyCheckpointAborted(long checkpointId) {
if (checkpointId == this.checkpointId) { if (checkpointId == this.checkpointId && !WriteMetadataEvent.BOOTSTRAP_INSTANT.equals(this.instant)) {
executor.execute(() -> { executor.execute(() -> {
this.ckpMetadata.abortInstant(this.instant); this.ckpMetadata.abortInstant(this.instant);
}, "abort instant %s", this.instant); }, "abort instant %s", this.instant);

2
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java
View File

@@ -97,8 +97,6 @@ public class CkpMetadata implements Serializable {
public void bootstrap(HoodieTableMetaClient metaClient) throws IOException { public void bootstrap(HoodieTableMetaClient metaClient) throws IOException {
fs.delete(path, true); fs.delete(path, true);
fs.mkdirs(path); fs.mkdirs(path);
metaClient.getActiveTimeline().getCommitsTimeline().filterPendingExcludingCompaction()
.lastInstant().ifPresent(instant -> startInstant(instant.getTimestamp()));
} }
public void startInstant(String instant) { public void startInstant(String instant) {