Do not create heartbeat executor when using Netty

Author: acogoluegnes

src/main/java/com/rabbitmq/perf/MulticastParams.java

@@ -139,6 +139,7 @@ public class MulticastParams {
   private FunctionalLogger functionalLogger = FunctionalLogger.NO_OP;
 
   private PrintStream out = System.out;
+  private boolean netty = false;
 
   public void setExchangeType(String exchangeType) {
     this.exchangeType = exchangeType;
@@ -773,6 +774,14 @@ public void setProducerSchedulerThreadCount(int producerSchedulerThreadCount) {
     this.producerSchedulerThreadCount = producerSchedulerThreadCount;
   }
 
+  void setNetty(boolean netty) {
+    this.netty = netty;
+  }
+
+  boolean netty() {
+    return this.netty;
+  }
+
   /**
    * Contract to handle the creation and configuration of resources. E.g. creation of queues,
    * binding exchange to queues.

src/main/java/com/rabbitmq/perf/MulticastSet.java

@@ -236,11 +236,15 @@ public void run(boolean announceStartup)
             : params.getServersUpLimit(),
         uris,
         factory)) {
-      // TODO do not set a heartbeat executor if Netty is used
-      ScheduledExecutorService heartbeatSenderExecutorService =
-          this.threadingHandler.scheduledExecutorService(
-              "perf-test-heartbeat-sender-", this.params.getHeartbeatSenderThreads());
-      factory.setHeartbeatExecutor(heartbeatSenderExecutorService);
+      // heartbeat sender executor not necessary with Netty
+      if (!params.netty()) {
+        ScheduledExecutorService heartbeatSenderExecutorService =
+            this.threadingHandler.scheduledExecutorService(
+                "perf-test-heartbeat-sender-", this.params.getHeartbeatSenderThreads());
+        factory.setHeartbeatExecutor(heartbeatSenderExecutorService);
+        shutdownService.wrap(heartbeatSenderExecutorService::shutdownNow);
+      }
+
       // use a single-threaded executor for the configuration connection
       // this way, a default one is not created and this one will shut down
       // when the run ends.

src/main/java/com/rabbitmq/perf/PerfTest.java

@@ -174,7 +174,8 @@ && useDefaultSslContext(cmd, System.getProperties())) {
       factory.setRequestedFrameMax(frameMax);
       factory.setRequestedHeartbeat(heartbeat);
 
-      if (!netty(cmd)) {
+      boolean netty = netty(cmd);
+      if (netty) {
         factory.useBlockingIo();
       }
 
@@ -241,7 +242,7 @@ && useDefaultSslContext(cmd, System.getProperties())) {
         nioParams(factory).setSslEngineConfigurator(Utils.sslEngineConfigurator(cmd));
       }
 
-      if (netty(cmd)) {
+      if (netty) {
         factory.netty().channelCustomizer(Utils.channelCustomizer(cmd));
         if (factory.isSSL()) {
           factory.netty().sslContext(Utils.nettySslContext(sslContext));
@@ -682,6 +683,8 @@ static MulticastParams multicastParams(
       functionalLogger = new DefaultFunctionalLogger(perfTestOptions.consoleOut, verboseFull);
     }
 
+    boolean netty = netty(cmd);
+
     MulticastParams p = new MulticastParams();
     p.setAutoAck(autoAck);
     p.setAutoDelete(autoDelete);
@@ -748,6 +751,7 @@ static MulticastParams multicastParams(
     p.setRateLimiterFactory(rateLimiterFactory);
     p.setFunctionalLogger(functionalLogger);
     p.setOut(consoleOut);
+    p.setNetty(netty);
     return p;
   }
 

src/test/java/com/rabbitmq/perf/it/PertTestIT.java renamed to src/test/java/com/rabbitmq/perf/it/PerfTestIT.java

@@ -36,7 +36,7 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
-public class PertTestIT {
+public class PerfTestIT {
 
   static ExecutorService executor = Executors.newSingleThreadExecutor();
   volatile ByteArrayOutputStream out, err;