Add --consumer-latency

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/perf/Converters.java

@@ -304,6 +304,23 @@ public Duration convert(String value) {
     }
   }
 
+  static class MicroSecondsToDurationTypeConverter implements CommandLine.ITypeConverter<Duration> {
+
+    @Override
+    public Duration convert(String value) {
+      try {
+        Duration duration = Duration.ofNanos(Long.parseLong(value) * 1_000);
+        if (duration.isNegative()) {
+          throw new CommandLine.TypeConversionException(
+              "'" + value + "' is not valid, it must be greater than or equal to 0");
+        }
+        return duration;
+      } catch (NumberFormatException e) {
+        throw new CommandLine.TypeConversionException("'" + value + "' is not a valid number");
+      }
+    }
+  }
+
   static class LeaderLocatorTypeConverter
       implements CommandLine.ITypeConverter<StreamCreator.LeaderLocator> {
 

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -556,6 +556,13 @@ static class InstanceSyncOptions {
       defaultValue = "true")
   private boolean tcpNoDelay;
 
+  @CommandLine.Option(
+      names = {"--consumer-latency", "-L"},
+      description = "consumer latency in microseconds",
+      converter = Converters.MicroSecondsToDurationTypeConverter.class,
+      defaultValue = "0")
+  private Duration consumerLatency;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
   private List<Monitoring> monitorings;
@@ -1124,6 +1131,7 @@ public Integer call() throws Exception {
                                   .builder();
                         }
 
+                        Runnable latencyWorker = Utils.latencyWorker(this.consumerLatency);
                         consumerBuilder =
                             consumerBuilder.messageHandler(
                                 (context, message) -> {
@@ -1137,12 +1145,12 @@ public Integer call() throws Exception {
                                     // tool
                                   }
                                   metrics.offset(context.offset());
+                                  latencyWorker.run();
                                 });
 
                         consumerBuilder = maybeConfigureForFiltering(consumerBuilder);
 
-                        Consumer consumer = consumerBuilder.build();
-                        return consumer;
+                        return consumerBuilder.build();
                       })
                   .collect(Collectors.toList()));
 

src/main/java/com/rabbitmq/stream/perf/Utils.java

@@ -481,4 +481,30 @@ static int filteringSubSetSize(int setSize) {
       return setSize - 3;
     }
   }
+
+  static Runnable latencyWorker(Duration latency) {
+    if (latency.isZero()) {
+      return () -> {};
+    } else if (latency.toMillis() >= 1) {
+      long latencyInMs = latency.toMillis();
+      return () -> latencySleep(latencyInMs);
+    } else {
+      long latencyInNs = latency.toNanos();
+      return () -> latencyBusyWait(latencyInNs);
+    }
+  }
+
+  private static void latencySleep(long delayInMs) {
+    try {
+      Thread.sleep(delayInMs);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  private static void latencyBusyWait(long delayInNs) {
+    long start = System.nanoTime();
+    while (System.nanoTime() - start < delayInNs)
+      ;
+  }
 }

src/test/java/com/rabbitmq/stream/perf/ConvertersTest.java

@@ -207,6 +207,26 @@ void sniServerNamesConverter() {
         .contains(new SNIHostName("dummy"));
   }
 
+  @ParameterizedTest
+  @CsvSource({"5000,5", "0,0", "1000,1"})
+  void microSecondsToDurationTypeConverterOk(String value, long expectedInMs) {
+    Converters.MicroSecondsToDurationTypeConverter converter =
+        new Converters.MicroSecondsToDurationTypeConverter();
+    Duration duration = converter.convert(value);
+    assertThat(duration).isNotNull();
+    assertThat(duration).isEqualTo(Duration.ofMillis(expectedInMs));
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"-1000000", "abc", "1.5"})
+  void microSecondsToDurationTypeConverterKo(String value) {
+    Converters.MicroSecondsToDurationTypeConverter converter =
+        new Converters.MicroSecondsToDurationTypeConverter();
+    assertThatThrownBy(() -> converter.convert(value))
+        .isInstanceOf(CommandLine.TypeConversionException.class)
+        .hasMessageContaining("valid");
+  }
+
   private static Tag tag(String key, String value) {
     return Tag.of(key, value);
   }