lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4450] Revert the checkpoint abort notification (#6181)

Browse Source
This commit is contained in:
Danny Chan
2022-07-24 08:44:22 +08:00
committed by GitHub
parent a0ffd05b77
commit ba11082282
2 changed files with 1 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -267,15 +267,6 @@ public class StreamWriteOperatorCoordinator
);
}
@Override
public void notifyCheckpointAborted(long checkpointId) {
if (checkpointId == this.checkpointId && !WriteMetadataEvent.BOOTSTRAP_INSTANT.equals(this.instant)) {
executor.execute(() -> {
this.ckpMetadata.abortInstant(this.instant);
}, "abort instant %s", this.instant);
}
}
@Override
public void resetToCheckpoint(long checkpointID, byte[] checkpointData) {
// no operation

9
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java
View File

@@ -243,13 +243,6 @@ public abstract class AbstractStreamWriteFunction<I>
return this.ckpMetadata.lastPendingInstant();
}
/**
* Returns whether the instant is fresh new(not aborted).
*/
protected boolean freshInstant(String instant) {
return !this.ckpMetadata.isAborted(instant);
}
/**
* Prepares the instant time to write with for next checkpoint.
*
@@ -286,6 +279,6 @@ public abstract class AbstractStreamWriteFunction<I>
* Returns whether the pending instant is invalid to write with.
*/
private boolean invalidInstant(String instant, boolean hasData) {
return instant.equals(this.currentInstant) && hasData && freshInstant(instant);
return instant.equals(this.currentInstant) && hasData;
}
}