lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[MINOR] Added log to debug checkpoint resumption when set to 0 (#4650)

Browse Source
This commit is contained in:
Harsha Teja Kanna
2022-01-28 22:08:25 -06:00
committed by GitHub
parent c0e8b03d93
commit ed7aa138e8
1 changed files with 2 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
View File

@@ -370,6 +370,7 @@ public class DeltaSync implements Serializable {
.initTable(new Configuration(jssc.hadoopConfiguration()), cfg.targetBasePath); .initTable(new Configuration(jssc.hadoopConfiguration()), cfg.targetBasePath);
} }
LOG.debug("Checkpoint from config: " + cfg.checkpoint);
if (!resumeCheckpointStr.isPresent() && cfg.checkpoint != null) { if (!resumeCheckpointStr.isPresent() && cfg.checkpoint != null) {
resumeCheckpointStr = Option.of(cfg.checkpoint); resumeCheckpointStr = Option.of(cfg.checkpoint);
} }
@@ -470,6 +471,7 @@ public class DeltaSync implements Serializable {
Option<HoodieCommitMetadata> commitMetadataOption = getLatestCommitMetadataWithValidCheckpointInfo(commitTimelineOpt.get()); Option<HoodieCommitMetadata> commitMetadataOption = getLatestCommitMetadataWithValidCheckpointInfo(commitTimelineOpt.get());
if (commitMetadataOption.isPresent()) { if (commitMetadataOption.isPresent()) {
HoodieCommitMetadata commitMetadata = commitMetadataOption.get(); HoodieCommitMetadata commitMetadata = commitMetadataOption.get();
LOG.debug("Checkpoint reset from metadata: " + commitMetadata.getMetadata(CHECKPOINT_RESET_KEY));
if (cfg.checkpoint != null && (StringUtils.isNullOrEmpty(commitMetadata.getMetadata(CHECKPOINT_RESET_KEY)) if (cfg.checkpoint != null && (StringUtils.isNullOrEmpty(commitMetadata.getMetadata(CHECKPOINT_RESET_KEY))
|| !cfg.checkpoint.equals(commitMetadata.getMetadata(CHECKPOINT_RESET_KEY)))) { || !cfg.checkpoint.equals(commitMetadata.getMetadata(CHECKPOINT_RESET_KEY)))) {
resumeCheckpointStr = Option.of(cfg.checkpoint); resumeCheckpointStr = Option.of(cfg.checkpoint);