From 36a9f63e45b97a37df7e53691625172e2934b605 Mon Sep 17 00:00:00 2001
From: Danny Chan <yuzhao.cyz@gmail.com>
Date: Mon, 17 Jan 2022 18:18:45 +0800
Subject: [PATCH] [HUDI-3257] Excluding clustering instants from pending
 rollback info (#4616)

---
 .../hudi/client/AbstractHoodieWriteClient.java       | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
index 66c76ffbc..d4315202e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
@@ -906,13 +906,17 @@ public abstract class AbstractHoodieWriteClient<T extends HoodieRecordPayload, I
   protected Map<String, Option<HoodiePendingRollbackInfo>> getPendingRollbackInfos(HoodieTableMetaClient metaClient) {
     List<HoodieInstant> instants = metaClient.getActiveTimeline().filterPendingRollbackTimeline().getInstants().collect(Collectors.toList());
     Map<String, Option<HoodiePendingRollbackInfo>> infoMap = new HashMap<>();
-    HoodieTimeline pendingCompactionTimeline = metaClient.getActiveTimeline().filterPendingCompactionTimeline();
     for (HoodieInstant instant : instants) {
       try {
         HoodieRollbackPlan rollbackPlan = RollbackUtils.getRollbackPlan(metaClient, instant);
-        String instantToRollback = rollbackPlan.getInstantToRollback().getCommitTime();
-        if (!pendingCompactionTimeline.containsInstant(instantToRollback)) {
-          infoMap.putIfAbsent(instantToRollback, Option.of(new HoodiePendingRollbackInfo(instant, rollbackPlan)));
+        String action = rollbackPlan.getInstantToRollback().getAction();
+        if (!HoodieTimeline.COMPACTION_ACTION.equals(action)) {
+          boolean isClustering = HoodieTimeline.REPLACE_COMMIT_ACTION.equals(action)
+              && ClusteringUtils.getClusteringPlan(metaClient, instant).isPresent();
+          if (!isClustering) {
+            String instantToRollback = rollbackPlan.getInstantToRollback().getCommitTime();
+            infoMap.putIfAbsent(instantToRollback, Option.of(new HoodiePendingRollbackInfo(instant, rollbackPlan)));
+          }
         }
       } catch (IOException e) {
         LOG.warn("Fetching rollback plan failed for " + infoMap + ", skip the plan", e);