HBASE-27563 ChaosMonkey sometimes generates invalid boundaries for ra…

…ndom item selection

Signed-off-by: Duo Zhang <zhangduo@apache.org>

hbase-it/src/test/java/org/apache/hadoop/hbase/chaos/monkies/PolicyBasedChaosMonkey.java

@@ -30,13 +30,17 @@
 import org.apache.hadoop.hbase.IntegrationTestingUtility;
 import org.apache.hadoop.hbase.chaos.policies.Policy;
 import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.hbase.util.ReservoirSample;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 /**
  * Chaos monkey that given multiple policies will run actions against the cluster.
  */
 public class PolicyBasedChaosMonkey extends ChaosMonkey {
+  private static final Logger LOG = LoggerFactory.getLogger(PolicyBasedChaosMonkey.class);
 
   private static final long ONE_SEC = 1000;
   private static final long ONE_MIN = 60 * ONE_SEC;
@@ -116,13 +120,14 @@ public static <T> T selectWeightedRandomItem(List<Pair<T, Integer>> items) {
 
   /** Selects and returns ceil(ratio * items.length) random items from the given array */
   public static <T> List<T> selectRandomItems(T[] items, float ratio) {
-    int selectedNumber = (int) Math.ceil(items.length * ratio);
-
-    List<T> originalItems = Arrays.asList(items);
-    Collections.shuffle(originalItems);
-
-    int startIndex = ThreadLocalRandom.current().nextInt(items.length - selectedNumber);
-    return originalItems.subList(startIndex, startIndex + selectedNumber);
+    // clamp ratio to [0.0,1.0]
+    ratio = Math.max(Math.min(ratio, 1.0f), 0.0f);
+    final int selectedNumber = (int) Math.ceil(items.length * ratio);
+    final ReservoirSample<T> sample = new ReservoirSample<>(selectedNumber);
+    sample.add(Arrays.stream(items));
+    final List<T> shuffledItems = sample.getSamplingResult();
+    Collections.shuffle(shuffledItems);
+    return shuffledItems;
   }
 
   @Override
@@ -151,7 +156,10 @@ public boolean isStopped() {
 
   @Override
   public void waitForStop() throws InterruptedException {
-    monkeyThreadPool.awaitTermination(1, TimeUnit.MINUTES);
+    if (!monkeyThreadPool.awaitTermination(1, TimeUnit.MINUTES)) {
+      LOG.warn("Some pool threads failed to terminate. Forcing. {}", monkeyThreadPool);
+      monkeyThreadPool.shutdownNow();
+    }
   }
 
   @Override