[HUDI-4152] Add IT for InstantCompactionPlanSelectStrategy

Author: lanyuanxiaoyao

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/strategy/InstantCompactionPlanSelectStrategy.java

@@ -21,11 +21,10 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.stream.Stream;
+import java.util.stream.Collectors;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.StringUtils;
-import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.sink.compact.FlinkCompactionConfig;
 import org.apache.hudi.sink.compact.HoodieFlinkCompactor;
 import org.slf4j.Logger;
@@ -43,11 +42,9 @@ public List<HoodieInstant> select(HoodieTimeline pendingCompactionTimeline, Flin
       LOG.warn("None instant is selected");
       return Collections.emptyList();
     }
-    Stream<String> instants = Arrays.stream(config.compactionPlanInstant.split(","));
-    HoodieInstant specifiedInstant = pendingCompactionTimeline.getInstants()
-        .filter(instant -> instants.anyMatch(i -> i.equals(instant.getTimestamp())))
-        .findFirst()
-        .orElseThrow(() -> new HoodieException("The instant " + config.compactionPlanInstant + " is not found in timeline"));
-    return Collections.singletonList(specifiedInstant);
+    List<String> instants = Arrays.asList(config.compactionPlanInstant.split(","));
+    return pendingCompactionTimeline.getInstants()
+        .filter(instant -> instants.contains(instant.getTimestamp()))
+        .collect(Collectors.toList());
   }
 }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/strategy/MultiCompactionPlanSelectStrategy.java

@@ -33,7 +33,7 @@ public class MultiCompactionPlanSelectStrategy implements CompactionPlanSelectSt
   @Override
   public List<HoodieInstant> select(HoodieTimeline pendingCompactionTimeline, FlinkCompactionConfig config) {
     List<HoodieInstant> pendingCompactionPlanInstants = pendingCompactionTimeline.getInstants().collect(Collectors.toList());
-    if (!CompactionUtil.isLIFO(config.compactionSeq)) {
+    if (CompactionUtil.isLIFO(config.compactionSeq)) {
       Collections.reverse(pendingCompactionPlanInstants);
     }
     int range = Math.min(config.compactionPlanMaxSelect, pendingCompactionPlanInstants.size());

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/strategy/SingleCompactionPlanSelectStrategy.java

@@ -33,8 +33,8 @@ public class SingleCompactionPlanSelectStrategy implements CompactionPlanSelectS
   @Override
   public List<HoodieInstant> select(HoodieTimeline pendingCompactionTimeline, FlinkCompactionConfig config) {
     Option<HoodieInstant> compactionPlanInstant = CompactionUtil.isLIFO(config.compactionSeq)
-        ? pendingCompactionTimeline.firstInstant()
-        : pendingCompactionTimeline.lastInstant();
+        ? pendingCompactionTimeline.lastInstant()
+        : pendingCompactionTimeline.firstInstant();
     if (compactionPlanInstant.isPresent()) {
       return Collections.singletonList(compactionPlanInstant.get());
     }

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/TestCompactionPlanSelectStrategy.java

@@ -119,8 +119,14 @@ void testInstantCompactionPlanSelectStrategy() {
     InstantCompactionPlanSelectStrategy strategy = new InstantCompactionPlanSelectStrategy();
     assertHoodieInstantsEquals(new HoodieInstant[]{INSTANT_004}, strategy.select(pendingCompactionTimeline, compactionConfig));
 
+    compactionConfig.compactionPlanInstant = "002,003";
+    assertHoodieInstantsEquals(new HoodieInstant[]{INSTANT_002, INSTANT_003}, strategy.select(pendingCompactionTimeline, compactionConfig));
+
+    compactionConfig.compactionPlanInstant = "002,005";
+    assertHoodieInstantsEquals(new HoodieInstant[]{INSTANT_002}, strategy.select(pendingCompactionTimeline, compactionConfig));
+
     compactionConfig.compactionPlanInstant = "005";
-    assertThrows(HoodieException.class, () -> strategy.select(pendingCompactionTimeline, compactionConfig));
+    assertHoodieInstantsEquals(new HoodieInstant[]{}, strategy.select(pendingCompactionTimeline, compactionConfig));
   }
 
   private void assertHoodieInstantsEquals(HoodieInstant[] expected, List<HoodieInstant> actual) {