hudi-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] [incubator-hudi] pratyakshsharma commented on a change in pull request #1362: HUDI-644 Enable user to get checkpoint from previous commits in DeltaStreamer
Date Mon, 02 Mar 2020 09:35:26 GMT
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_r386281673
 
 

 ##########
 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:
   Here you are trying to iterate from the very beginning, what about the case where commit
2 and commit 5 are done using DeltaStreamer and both of them have CHECKPOINT_KEY. In this
case commit 5 has the latest checkpoint to use but as per your logic, you end up considering
commit 2 checkpoint. 

----------------------------------------------------------------
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:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message