Add support for Netty

This commit adds support for the Netty-based frame handler that was
added in RabbitMQ AMQP 091 Java client 5.27.0.

New options:
 * --netty: to activate Netty with the defaults
 * --netty-threads: to set the number of threads used in the event loop
 * --netty-epoll: to use the native epoll support (Linux x86-64)
 * --netty-kqueue: to use the native kqueue support (macOS aarch64)

This change also deprecates the options related to Java NIO.

References rabbitmq/rabbitmq-java-client#1663
Fixes #856

Author: acogoluegnes

.github/workflows/test-alphas.yml

@@ -19,6 +19,11 @@ jobs:
     name:  Test against ${{ matrix.rabbitmq-image }}
     steps:
       - uses: actions/checkout@v4
+      - name: Checkout tls-gen
+        uses: actions/checkout@v4
+        with:
+          repository: rabbitmq/tls-gen
+          path: './tls-gen'
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:

.github/workflows/test-pr.yml

@@ -12,6 +12,11 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
+      - name: Checkout tls-gen
+        uses: actions/checkout@v4
+        with:
+          repository: rabbitmq/tls-gen
+          path: './tls-gen'
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:

.github/workflows/test-supported-java-versions.yml

@@ -18,6 +18,11 @@ jobs:
     name:  Test against Java ${{ matrix.distribution }} ${{ matrix.version }}
     steps:
       - uses: actions/checkout@v4
+      - name: Checkout tls-gen
+        uses: actions/checkout@v4
+        with:
+          repository: rabbitmq/tls-gen
+          path: './tls-gen'
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:

.github/workflows/test.yml

@@ -12,6 +12,11 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
+      - name: Checkout tls-gen
+        uses: actions/checkout@v4
+        with:
+          repository: rabbitmq/tls-gen
+          path: './tls-gen'
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:

ci/start-broker.sh

@@ -10,11 +10,32 @@ wait_for_message() {
   done
 }
 
+make -C "${PWD}"/tls-gen/basic
+
+mkdir -p rabbitmq-configuration/tls
+cp -R "${PWD}"/tls-gen/basic/result/* rabbitmq-configuration/tls
+chmod o+r rabbitmq-configuration/tls/*
+chmod g+r rabbitmq-configuration/tls/*
+
+echo "loopback_users = none
+
+listeners.ssl.default = 5671
+
+ssl_options.cacertfile = /etc/rabbitmq/tls/ca_certificate.pem
+ssl_options.certfile   = /etc/rabbitmq/tls/server_$(hostname)_certificate.pem
+ssl_options.keyfile    = /etc/rabbitmq/tls/server_$(hostname)_key.pem
+ssl_options.verify     = verify_peer
+ssl_options.fail_if_no_peer_cert = false
+ssl_options.depth = 1
+
+auth_mechanisms.1 = PLAIN" >> rabbitmq-configuration/rabbitmq.conf
+
 echo "Running RabbitMQ ${RABBITMQ_IMAGE}"
 
 docker rm -f rabbitmq 2>/dev/null || echo "rabbitmq was not running"
 docker run -d --name rabbitmq \
     --network host \
+    -v "${PWD}"/rabbitmq-configuration:/etc/rabbitmq \
     "${RABBITMQ_IMAGE}"
 
 wait_for_message rabbitmq "completed with"

pom.xml

@@ -55,13 +55,14 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
     <spotless.check.skip>true</spotless.check.skip>
-    <rabbitmq.version>5.26.0</rabbitmq.version>
+    <rabbitmq.version>5.27.0-SNAPSHOT</rabbitmq.version>
     <slf4j.version>2.0.17</slf4j.version>
     <commons-cli.version>1.10.0</commons-cli.version>
     <metrics.version>4.2.33</metrics.version>
     <micrometer.version>1.15.2</micrometer.version>
     <jgroups.version>5.4.8.Final</jgroups.version>
     <jgroups-kubernetes.version>2.0.2.Final</jgroups-kubernetes.version>
+    <netty.version>4.2.3.Final</netty.version>
     <gson.version>2.13.1</gson.version>
     <resilience4j.version>2.1.0</resilience4j.version>
     <logback.version>1.3.15</logback.version>
@@ -169,6 +170,18 @@
       <artifactId>jgroups-kubernetes</artifactId>
       <version>${jgroups-kubernetes.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-epoll</artifactId>
+      <version>${netty.version}</version>
+      <classifier>linux-x86_64</classifier>
+    </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-kqueue</artifactId>
+      <version>${netty.version}</version>
+      <classifier>osx-aarch_64</classifier>
+    </dependency>
 
     <dependency>
       <groupId>org.junit.jupiter</groupId>

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

@@ -25,6 +25,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.IOException;
+import java.io.PrintStream;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.Callable;
@@ -98,6 +99,7 @@ public class Consumer extends AgentBase implements Runnable {
 
   private final Runnable rateLimiterCallback;
   private final boolean rateLimitation;
+  private final PrintStream out;
 
   public Consumer(ConsumerParameters parameters) {
     super(
@@ -124,6 +126,7 @@ public Consumer(ConsumerParameters parameters) {
 
     this.queueNames.set(new ArrayList<>(parameters.getQueueNames()));
     this.initialQueueNames = new ArrayList<>(parameters.getQueueNames());
+    this.out = parameters.getOut();
 
     if (parameters.getConsumerLatenciesIndicator().isVariable()) {
       this.consumerLatency =
@@ -372,7 +375,7 @@ public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig
 
     @Override
     public void handleCancel(String consumerTag) {
-      System.out.printf("Consumer cancelled by broker for tag: %s\n", consumerTag);
+      out.printf("Consumer cancelled by broker for tag: %s\n", consumerTag);
       epochMessageCount.set(0);
       if (consumerTagBranchMap.containsKey(consumerTag)) {
         String qName = consumerTagBranchMap.get(consumerTag);
@@ -393,7 +396,7 @@ public void handleCancel(String consumerTag) {
             delay.toMillis(),
             TimeUnit.MILLISECONDS);
       } else {
-        System.out.printf("Could not find queue for consumer tag: %s\n", consumerTag);
+        out.printf("Could not find queue for consumer tag: %s\n", consumerTag);
       }
     }
   }

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

@@ -18,6 +18,7 @@
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
 import com.rabbitmq.perf.metrics.PerformanceMetrics;
+import java.io.PrintStream;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
@@ -60,6 +61,16 @@ public class ConsumerParameters {
 
   private int id;
   private FunctionalLogger functionalLogger = FunctionalLogger.NO_OP;
+  private PrintStream out = System.out;
+
+  ConsumerParameters setOut(PrintStream out) {
+    this.out = out;
+    return this;
+  }
+
+  PrintStream getOut() {
+    return out;
+  }
 
   public Channel getChannel() {
     return channel;

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

@@ -24,6 +24,7 @@
 import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
 import com.rabbitmq.perf.metrics.PerformanceMetrics;
 import java.io.IOException;
+import java.io.PrintStream;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -137,6 +138,9 @@ 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;
   }
@@ -319,6 +323,10 @@ void setConsumerStartDelay(Duration csd) {
     this.consumerStartDelay = csd;
   }
 
+  void setOut(PrintStream out) {
+    this.out = out;
+  }
+
   public int getConsumerCount() {
     return consumerCount;
   }
@@ -485,6 +493,10 @@ public Duration getConsumerStartDelay() {
     return consumerStartDelay;
   }
 
+  PrintStream getOut() {
+    return out;
+  }
+
   public void setPolling(boolean polling) {
     this.polling = polling;
   }
@@ -655,7 +667,8 @@ public Consumer createConsumer(
                     topologyRecordingScheduledExecutorService)
                 .setStartListener(this.startListener)
                 .setRateLimiterFactory(this.rateLimiterFactory)
-                .setFunctionalLogger(this.functionalLogger));
+                .setFunctionalLogger(this.functionalLogger)
+                .setOut(this.out));
     this.topologyHandler.next();
     return consumer;
   }
@@ -761,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

@@ -28,6 +28,7 @@
 import com.rabbitmq.perf.metrics.PerformanceMetrics;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import java.io.IOException;
+import java.io.PrintStream;
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
@@ -80,6 +81,7 @@ public class MulticastSet {
   private final ConnectionCreator connectionCreator;
   private final ExpectedMetrics expectedMetrics;
   private final InstanceSynchronization instanceSynchronization;
+  private final PrintStream out;
 
   public MulticastSet(
       PerformanceMetrics performanceMetrics,
@@ -167,6 +169,7 @@ public MulticastSet(
     this.connectionCreator = new ConnectionCreator(this.factory, this.uris, connectionAllocation);
     this.expectedMetrics = expectedMetrics;
     this.instanceSynchronization = instanceSynchronization;
+    this.out = params.getOut();
   }
 
   protected static int nbThreadsForConsumer(MulticastParams params) {
@@ -233,10 +236,17 @@ public void run(boolean announceStartup)
             : params.getServersUpLimit(),
         uris,
         factory)) {
-      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);
+        if (heartbeatSenderExecutorService != null) {
+          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.
@@ -382,7 +392,7 @@ public void run(boolean announceStartup)
 
       executeShutdownSequence.run();
     } else {
-      System.out.println(
+      out.println(
           "Could not connect to broker(s) in "
               + params.getServersStartUpTimeout()
               + " second(s), exiting.");
@@ -475,7 +485,7 @@ private void createConsumers(
     int consumerIndex = 0;
     for (int i = 0; i < consumerConnections.length; i++) {
       if (announceStartup) {
-        System.out.println("id: " + testID + ", starting consumer #" + i);
+        out.println("id: " + testID + ", starting consumer #" + i);
       }
       ExecutorService executorService = consumersExecutorsFactory.apply(i);
       factory.setSharedExecutor(executorService);
@@ -484,7 +494,7 @@ private void createConsumers(
       consumerConnections[i] = consumerConnection;
       for (int j = 0; j < params.getConsumerChannelCount(); j++) {
         if (announceStartup) {
-          System.out.println("id: " + testID + ", starting consumer #" + i + ", channel #" + j);
+          out.println("id: " + testID + ", starting consumer #" + i + ", channel #" + j);
         }
         Consumer consumer =
             params.createConsumer(
@@ -507,13 +517,13 @@ private void createProducers(
     int producerIndex = 0;
     for (int i = 0; i < producerConnections.length; i++) {
       if (announceStartup) {
-        System.out.println("id: " + testID + ", starting producer #" + i);
+        out.println("id: " + testID + ", starting producer #" + i);
       }
       Connection producerConnection = createConnection(PRODUCER_THREAD_PREFIX + i);
       producerConnections[i] = producerConnection;
       for (int j = 0; j < params.getProducerChannelCount(); j++) {
         if (announceStartup) {
-          System.out.println("id: " + testID + ", starting producer #" + i + ", channel #" + j);
+          out.println("id: " + testID + ", starting producer #" + i + ", channel #" + j);
         }
         AgentState agentState = new AgentState();
         agentState.runnable =
@@ -538,7 +548,7 @@ private void startConsumers(Runnable[] consumerRunnables) throws InterruptedExce
         runnable.run();
         LOGGER.debug("Consumer runnable started");
         if (params.getConsumerSlowStart()) {
-          System.out.println("Delaying start by 1 second because -S/--slow-start was requested");
+          out.println("Delaying start by 1 second because -S/--slow-start was requested");
           Thread.sleep(1000);
         }
       }
@@ -551,8 +561,7 @@ private void startConsumers(Runnable[] consumerRunnables) throws InterruptedExce
                 for (Runnable runnable : consumerRunnables) {
                   runnable.run();
                   if (params.getConsumerSlowStart()) {
-                    System.out.println(
-                        "Delaying start by 1 second because -S/--slow-start was requested");
+                    out.println("Delaying start by 1 second because -S/--slow-start was requested");
                     try {
                       Thread.sleep(1000);
                     } catch (InterruptedException e) {