Repository: skiptests/astraea
Pull request #573 (2022-08-10 18:29:00)

Add simple chaos monkey to kill consumers
新增一隻簡單的猴子，該猴子會定時隨機重啟consumer，如此我們就可以觀察consumer balance的狀況

chia7712: @harryteng9527 麻煩看一下這隻PR，是否有符合目前的測試需求
harryteng9527: 看起來有，透過unsubscribe發送`leave group`觸發rebalance，等待`chaos duration`時間後重啟該consumer，再一次觸發rebalance

chia7712: > 看起來有，透過unsubscribe發送leave group觸發rebalance，等待chaos duration時間後重啟該consumer，再一次觸發rebalance

看看是否能更準確的觀察到讀取端的吞吐量變化

Patch:
diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
index 3123acd7..b205055a 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -35,6 +35,7 @@ import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.Compression;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.argument.CompressionField;
+import org.astraea.app.argument.DurationField;
 import org.astraea.app.argument.NonEmptyStringField;
 import org.astraea.app.argument.NonNegativeShortField;
 import org.astraea.app.argument.PathField;
@@ -166,6 +167,20 @@ public class Performance {
     var fileWriterFuture =
         fileWriter.map(CompletableFuture::runAsync).orElse(CompletableFuture.completedFuture(null));
 
+    var chaos =
+        param.chaosDuration == null
+            ? CompletableFuture.completedFuture(null)
+            : CompletableFuture.runAsync(
+                () -> {
+                  while (!consumerThreads.stream().allMatch(AbstractThread::closed)) {
+                    var thread =
+                        consumerThreads.get((int) (Math.random() * consumerThreads.size()));
+                    thread.unsubscribe();
+                    Utils.sleep(param.chaosDuration);
+                    thread.resubscribe();
+                  }
+                });
+
     CompletableFuture.runAsync(
         () -> {
           producerThreads.forEach(AbstractThread::waitForDone);
@@ -189,6 +204,7 @@ public class Performance {
     consumerThreads.forEach(AbstractThread::waitForDone);
     tracker.waitForDone();
     fileWriterFuture.join();
+    chaos.join();
     return param.topic;
   }
 
@@ -346,5 +362,13 @@ public class Performance {
         description = "Output format for the report",
         converter = ReportFormat.ReportFormatConverter.class)
     ReportFormat reportFormat = ReportFormat.CSV;
+
+    @Parameter(
+        names = {"--chaos.frequency"},
+        description =
+            "time to run the chaos monkey. It will kill consumer arbitrarily. There is no monkey by default",
+        validateWith = DurationField.class,
+        converter = DurationField.class)
+    Duration chaosDuration = null;
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
index f61cfdcd..62a70cd9 100644
--- a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
@@ -17,6 +17,7 @@
 package org.astraea.app.performance;
 
 import com.beust.jcommander.ParameterException;
+import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.function.BiConsumer;
 import org.astraea.app.argument.Argument;
@@ -156,4 +157,13 @@ public class PerformanceTest extends RequireBrokerCluster {
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments14));
   }
+
+  @Test
+  void testChaosFrequency() {
+    var args =
+        Argument.parse(
+            new Performance.Argument(),
+            new String[] {"--bootstrap.servers", "localhost:9092", "--chaos.frequency", "10s"});
+    Assertions.assertEquals(Duration.ofSeconds(10), args.chaosDuration);
+  }
 }