[MINOR] Avoid wrong way to get the latest completed instant (apache#1…

…2590) 1. avoid wrong way to get the latest completed instant Signed-off-by: TheR1sing3un <[email protected]>
acryldata · Jan 8, 2025 · 4efc241 · 4efc241
1 parent 5d888eb
commit 4efc241
Show file tree

Hide file tree

Showing 6 changed files with 6 additions and 27 deletions.
diff --git a/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java b/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
@@ -896,11 +896,7 @@ public void updateLastCommitTimeSynced(String tableName) {
     HoodieTimeline activeTimeline = getActiveTimeline();
     Option<String> lastCommitSynced = activeTimeline.lastInstant().map(HoodieInstant::requestedTime);
     Option<String> lastCommitCompletionSynced = activeTimeline
-        .getInstantsOrderedByCompletionTime()
-        .skip(activeTimeline.countInstants() - 1)
-        .findFirst()
-        .map(i -> Option.of(i.getCompletionTime()))
-        .orElse(Option.empty());
+        .getLatestCompletionTime();
     if (lastCommitSynced.isPresent()) {
       try {
         HashMap<String, String> propertyMap = new HashMap<>();

diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
@@ -405,7 +405,7 @@ public interface HoodieTimeline extends Serializable {
   Option<String> getLatestCompletionTime();
 
   /**
-   * Get the stream of instants in order by state transition timestamp of actions.
+   * Get the stream of instants in order by completion timestamp of actions.
    */
   Stream<HoodieInstant> getInstantsOrderedByCompletionTime();
 

diff --git a/...park-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamSourceV1.scala b/...park-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamSourceV1.scala
@@ -107,11 +107,7 @@ class HoodieStreamSourceV1(sqlContext: SQLContext,
     filteredTimeline match {
       case activeInstants if !activeInstants.empty() =>
         val timestamp = if (hollowCommitHandling == USE_TRANSITION_TIME) {
-          activeInstants.getInstantsOrderedByCompletionTime
-            .skip(activeInstants.countInstants() - 1)
-            .findFirst()
-            .get()
-            .getCompletionTime
+          activeInstants.getLatestCompletionTime.get()
         } else {
           activeInstants.lastInstant().get().requestedTime()
         }

diff --git a/...-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/DataHubSyncClient.java b/...-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/DataHubSyncClient.java
@@ -99,13 +99,7 @@ protected Option<String> getLastCommitTime() {
   }
 
   protected Option<String> getLastCommitCompletionTime() {
-    int countInstants = getActiveTimeline().countInstants();
-    return getActiveTimeline()
-        .getInstantsOrderedByCompletionTime()
-        .skip(countInstants - 1)
-        .findFirst()
-        .map(HoodieInstant::getCompletionTime)
-        .map(Option::of).orElseGet(Option::empty);
+    return getActiveTimeline().getLatestCompletionTime();
   }
 
   @Override

diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
@@ -404,12 +404,7 @@ public void updateLastCommitTimeSynced(String tableName) {
     // Set the last commit time and commit completion from the TBLproperties
     HoodieTimeline activeTimeline = getActiveTimeline();
     Option<String> lastCommitSynced = activeTimeline.lastInstant().map(HoodieInstant::requestedTime);
-    Option<String> lastCommitCompletionSynced = activeTimeline
-        .getInstantsOrderedByCompletionTime()
-        .skip(activeTimeline.countInstants() - 1)
-        .findFirst()
-        .map(i -> Option.of(i.getCompletionTime()))
-        .orElse(Option.empty());
+    Option<String> lastCommitCompletionSynced = activeTimeline.getLatestCompletionTime();
     if (lastCommitSynced.isPresent()) {
       try {
         Table table = client.getTable(databaseName, tableName);

diff --git a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -1979,9 +1979,7 @@ private void reSyncHiveTable() {
   }
 
   private String getLastCommitCompletionTimeSynced() {
-    return hiveClient.getActiveTimeline()
-        .getInstantsOrderedByCompletionTime()
-        .skip(hiveClient.getActiveTimeline().countInstants() - 1).findFirst().get().getCompletionTime();
+    return hiveClient.getActiveTimeline().getLatestCompletionTime().get();
   }
 
   private void reInitHiveSyncClient() {