address comments

Author: wangxianghu

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

@@ -20,6 +20,7 @@
 
 import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
 import org.apache.hudi.common.util.PartitionPathEncodeUtils;
+import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.config.HoodieTTLConfig;
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
 import org.apache.hudi.table.HoodieTable;
@@ -34,6 +35,7 @@
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeParseException;
 import java.time.temporal.TemporalAccessor;
+import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -158,16 +160,38 @@ protected List<String> getExpiredPartitionsForTimeStrategy(List<String> partitio
     long cutoffMillis = resolveCutoffMillis(instantTime, ttlInMilis);
     DateTimeFormatter formatter = DateTimeFormatter.ofPattern(eventTimeFormat).withZone(ZoneOffset.UTC);
     int segCount = segmentCount(eventTimeFormat);
-    return partitionPathsForTTL.stream().parallel()
+
+    // Step 1: filter by event time. Pure computation over immutable inputs (no shared state, no
+    // I/O), so a raw parallel stream on the common ForkJoinPool is safe. In a healthy table the
+    // vast majority of partitions survive this filter, so most of the work stays here and never
+    // reaches the more expensive step 2.
+    List<String> eventTimeExpired = 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)
+            path, formatter, timeSegStartIndex, segCount, cutoffMillis,
+            shouldDeleteHiveDefaultPartition, hiveStylePartitioning))
+        .collect(Collectors.toList());
+    if (eventTimeExpired.isEmpty()) {
+      return Collections.emptyList();
+    }
+
+    // Step 2: drop partitions that an earlier TTL replace already emptied. Their event time is
+    // derived from the (unchanged) path, so without this filter they 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, mirroring KeepByTimeStrategy which keys off
+    // the surviving slices' last commit time and thus naturally skips deleted ones.
+    //
+    // hasLiveFileSlice touches hoodieTable.getHoodieView(), which for non-default view types
+    // (SPILLABLE_DISK, EMBEDDED_KV_STORE) mutates a backing store during first-time partition
+    // load under a shared read lock -- unsafe under unbounded concurrent access on the common
+    // ForkJoinPool. Route the lookup through the engine context with bounded parallelism, the
+    // same pattern KeepByTimeStrategy#getLastCommitTimeForPartitions uses, so that blocking file
+    // listings don't run on the common pool and respect getPartitionTTLStatsMaxParallelism.
+    int statsParallelism = Math.min(eventTimeExpired.size(), writeConfig.getPartitionTTLStatsMaxParallelism());
+    return hoodieTable.getContext().map(eventTimeExpired,
+        partitionPath -> Pair.of(partitionPath, hasLiveFileSlice(partitionPath)),
+        statsParallelism).stream()
+        .filter(Pair::getRight)
+        .map(Pair::getLeft)
         .collect(Collectors.toList());
   }