Modify the logic of shouldThrowExceptionIfLocalViewBehind method

Author: dongkelun

hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java

@@ -21,6 +21,7 @@
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.metrics.Registry;
 import org.apache.hudi.common.table.marker.MarkerOperation;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.table.timeline.dto.BaseFileDTO;
 import org.apache.hudi.common.table.timeline.dto.ClusteringOpDTO;
@@ -33,6 +34,7 @@
 import org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView;
 import org.apache.hudi.common.table.view.SyncableFileSystemView;
 import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.timeline.service.handlers.BaseFileHandler;
 import org.apache.hudi.timeline.service.handlers.FileSliceHandler;
@@ -502,14 +504,15 @@ public void handle(@NotNull Context context) throws Exception {
         if (refreshCheck) {
           long beginFinalCheck = System.currentTimeMillis();
           if (isLocalViewBehind(context)) {
-            String lastInstantTs = context.queryParam(RemoteHoodieTableFileSystemView.LAST_INSTANT_TS,
+            String lastKnownInstantFromClient = context.queryParam(RemoteHoodieTableFileSystemView.LAST_INSTANT_TS,
                 HoodieTimeline.INVALID_INSTANT_TS);
+            String timelineHashFromClient = context.queryParam(RemoteHoodieTableFileSystemView.TIMELINE_HASH, "");
             HoodieTimeline localTimeline =
                 viewManager.getFileSystemView(context.queryParam(RemoteHoodieTableFileSystemView.BASEPATH_PARAM)).getTimeline();
-            if (shouldThrowExceptionIfLocalViewBehind(localTimeline, lastInstantTs)) {
+            if (shouldThrowExceptionIfLocalViewBehind(localTimeline, timelineHashFromClient)) {
               String errMsg =
                       "Last known instant from client was "
-                              + lastInstantTs
+                              + lastKnownInstantFromClient
                               + " but server has the following timeline "
                               + localTimeline.getInstants().collect(Collectors.toList());
               throw new BadRequestResponse(errMsg);
@@ -547,12 +550,12 @@ public void handle(@NotNull Context context) throws Exception {
   /**
    * Determine whether to throw an exception when local view of table's timeline is behind that of client's view.
    */
-  private boolean shouldThrowExceptionIfLocalViewBehind(HoodieTimeline localTimeline, String lastInstantTs) {
-    HoodieTimeline afterLastInstantTimeLine = localTimeline.findInstantsAfter(lastInstantTs).filterCompletedInstants();
+  private boolean shouldThrowExceptionIfLocalViewBehind(HoodieTimeline localTimeline, String timelineHashFromClient) {
+    Option<HoodieInstant> lastInstant = localTimeline.lastInstant();
     // When performing async clean, we may have one more .clean.completed after lastInstantTs.
     // In this case, we do not need to throw an exception.
-    if (afterLastInstantTimeLine.countInstants() == 1
-          && afterLastInstantTimeLine.filter(s -> s.getAction().equals(HoodieTimeline.CLEAN_ACTION)).countInstants() == 1) {
+    if (lastInstant.isPresent() && lastInstant.get().getAction().equals(HoodieTimeline.CLEAN_ACTION)
+          && localTimeline.findInstantsBefore(lastInstant.get().getTimestamp()).getTimelineHash().equals(timelineHashFromClient)) {
       return false;
     } else {
       return true;