filter deleted partition

Author: wangxianghu

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/ttl/strategy/KeepByEventTimeStrategy.java

@@ -161,9 +161,29 @@ protected List<String> getExpiredPartitionsForTimeStrategy(List<String> partitio
     return partitionPathsForTTL.stream().parallel()
         .filter(path -> isPartitionExpiredByEventTime(
             path, formatter, timeSegStartIndex, segCount, cutoffMillis, shouldDeleteHiveDefaultPartition, hiveStylePartitioning))
+        // A partition emptied by a previous TTL replace commit keeps showing up in
+        // getAllPartitionPaths() until the cleaner physically removes it, and its event time is
+        // derived from the (unchanged) path, so it would be re-selected on every batch -- issuing
+        // an empty replace commit each time and never converging. Keep only partitions that still
+        // have a live file slice so the strategy is idempotent, mirroring KeepByTimeStrategy which
+        // keys off the surviving slices' last commit time and thus naturally skips deleted ones.
+        .filter(this::hasLiveFileSlice)
         .collect(Collectors.toList());
   }
 
+  /**
+   * Whether the partition still has at least one live file slice as of {@code instantTime}.
+   * File groups replaced by an earlier TTL delete are excluded by the file system view, so an
+   * already-deleted partition returns {@code false} here even while its directory lingers on
+   * storage awaiting the cleaner.
+   */
+  private boolean hasLiveFileSlice(String partitionPath) {
+    return hoodieTable.getHoodieView()
+        .getLatestFileSlicesBeforeOrOn(partitionPath, instantTime, true)
+        .findAny()
+        .isPresent();
+  }
+
   /**
    * Number of '/'-separated path segments the configured format occupies.
    * Example: {@code yyyy-MM-dd/HH} -> 2.

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestPartitionTTLManagement.java

@@ -185,6 +185,15 @@ public void testKeepByEventTime() {
       Assertions.assertEquals(Sets.newHashSet(partitions[0], partitions[1]),
           result.getPartitionToReplaceFileIds().keySet());
       Assertions.assertEquals(10, readRecords(partitions).size());
+
+      // Idempotency: the partitions we just replaced still linger in getAllPartitionPaths() until
+      // the cleaner runs, and their event time is derived from the (unchanged) path. Without the
+      // live-file-slice guard they would be re-selected here, producing an endless stream of empty
+      // replace commits. A second run must find nothing to delete.
+      String instantTime2 = client.startDeletePartitionCommit(metaClient);
+      HoodieWriteResult result2 = client.managePartitionTTL(instantTime2);
+      Assertions.assertTrue(result2.getPartitionToReplaceFileIds().isEmpty(),
+          "Already-deleted partitions must not be re-selected on the next TTL batch");
     }
   }