pratyakshsharma commented on a change in pull request #1362: HUDI-644 Enable
user to get checkpoint from previous commits in DeltaStreamer
URL: https://github.com/apache/incubator-hudi/pull/1362#discussion_r386845759
##########
File path:
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
##########
@@ -529,6 +514,47 @@ private void registerAvroSchemas(SchemaProvider
schemaProvider) {
}
}
+ /**
+ * Search delta streamer checkpoint from the previous commits.
+ *
+ * @param commitTimelineOpt HoodieTimeline object
+ * @return checkpoint metadata as String
+ */
+ private Option<String> retrieveCheckpointFromCommits(Option<HoodieTimeline>
commitTimelineOpt) throws Exception {
+ Option<HoodieInstant> lastCommit = commitTimelineOpt.get().lastInstant();
+ if (lastCommit.isPresent()) {
+ HoodieCommitMetadata commitMetadata = HoodieCommitMetadata
+
.fromBytes(commitTimelineOpt.get().getInstantDetails(lastCommit.get()).get(),
HoodieCommitMetadata.class);
+ // user-defined checkpoint appeared and not equal to the user-defined
checkpoint of the last commit
+ if (cfg.checkpoint != null &&
!cfg.checkpoint.equals(commitMetadata.getMetadata(CHECKPOINT_RESET_KEY))) {
+ return Option.of(cfg.checkpoint);
+ }
+ int commitsToCheckCnt;
+ // search check point from previous commits in backward
+ if (cfg.searchCheckpoint) {
+ commitsToCheckCnt = commitTimelineOpt.get().countInstants();
+ } else {
+ commitsToCheckCnt = 1; // only check the last commit
+ }
+ Option<HoodieInstant> curCommit;
+ for (int i = 0; i < commitsToCheckCnt; ++i) {
Review comment:
My bad. :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
