Fix epoll

Author: acogoluegnes

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

@@ -41,10 +41,12 @@
 import com.rabbitmq.perf.metrics.MetricsFormatterFactory;
 import com.rabbitmq.perf.metrics.MetricsFormatterFactory.Context;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.IoHandlerFactory;
 import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.channel.epoll.EpollIoHandler;
+import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.channel.nio.NioIoHandler;
 import java.io.*;
 import java.math.BigDecimal;
@@ -59,6 +61,7 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import javax.net.ssl.SSLContext;
@@ -180,8 +183,8 @@ && useDefaultSslContext(cmd, System.getProperties())) {
         factory.useBlockingIo();
       }
 
-      factory = configureNioIfRequested(cmd, factory, shutdownService);
-      factory = configureNettyIfRequested(cmd, factory, shutdownService);
+      configureNioIfRequested(cmd, factory, shutdownService);
+      configureNettyIfRequested(cmd, factory, shutdownService);
 
       String oauth2TokenEndpoint = strArg(cmd, "o2uri", null);
       if (oauth2TokenEndpoint != null) {
@@ -793,7 +796,7 @@ private static PrintWriter openCsvFileForWriting(
   }
 
   @SuppressWarnings("deprecation")
-  private static ConnectionFactory configureNioIfRequested(
+  private static void configureNioIfRequested(
       CommandLineProxy cmd, ConnectionFactory factory, ShutdownService shutdownService) {
     int nbThreads = Utils.intArg(cmd, "niot", -1);
     int executorSize = Utils.intArg(cmd, "niotp", -1);
@@ -822,7 +825,6 @@ private static ConnectionFactory configureNioIfRequested(
       factory.useNio();
       factory.setNioParams(nioParams);
     }
-    return factory;
   }
 
   protected static int[] getNioNbThreadsAndExecutorSize(
@@ -853,21 +855,22 @@ private static com.rabbitmq.client.impl.nio.NioParams nioParams(ConnectionFactor
     return cf.getNioParams();
   }
 
-  private static ConnectionFactory configureNettyIfRequested(
+  private static void configureNettyIfRequested(
       CommandLineProxy cmd, ConnectionFactory factory, ShutdownService shutdownService) {
     if (netty(cmd)) {
       int nbThreads = Utils.intArg(cmd, "ntyt", -1);
       boolean epoll = hasOption(cmd, "ntyep");
       IoHandlerFactory ioHandlerFactory =
           epoll ? EpollIoHandler.newFactory() : NioIoHandler.newFactory();
+      Consumer<Bootstrap> bootstrapCustomizer =
+          epoll ? b -> b.channel(EpollSocketChannel.class) : b -> {};
       EventLoopGroup eventLoopGroup =
           nbThreads > 0
               ? new MultiThreadIoEventLoopGroup(nbThreads, ioHandlerFactory)
               : new MultiThreadIoEventLoopGroup(ioHandlerFactory);
       shutdownService.wrap(() -> eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS));
-      factory.netty().eventLoopGroup(eventLoopGroup);
+      factory.netty().bootstrapCustomizer(bootstrapCustomizer).eventLoopGroup(eventLoopGroup);
     }
-    return factory;
   }
 
   private static boolean netty(CommandLineProxy cmd) {