Bump Netty to 4.2.0.RC3

Author: acogoluegnes

pom.xml

@@ -50,7 +50,7 @@
     <spotless.check.skip>true</spotless.check.skip>
     <slf4j.version>1.7.36</slf4j.version>
     <logback.version>1.2.13</logback.version>
-    <netty.version>4.1.119.Final</netty.version>
+    <netty.version>4.2.0.RC3</netty.version>
     <proton-j.version>0.34.1</proton-j.version>
     <metrics.version>4.2.30</metrics.version>
     <micrometer.version>1.14.5</micrometer.version>

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -59,17 +59,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufOutputStream;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.ChannelOutboundHandlerAdapter;
-import io.netty.channel.ChannelPromise;
-import io.netty.channel.ConnectTimeoutException;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.*;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.DecoderException;
@@ -248,7 +238,7 @@ public Client(ClientParameters parameters) {
     if (b.config().group() == null) {
       EventLoopGroup eventLoopGroup;
       if (parameters.eventLoopGroup == null) {
-        this.eventLoopGroup = new NioEventLoopGroup();
+        this.eventLoopGroup = Utils.eventLoopGroup();
         eventLoopGroup = this.eventLoopGroup;
       } else {
         this.eventLoopGroup = null;

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -38,7 +38,6 @@
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import java.io.IOException;
@@ -212,7 +211,7 @@ class StreamEnvironment implements Environment {
             this.addresses.size(), 1, "rabbitmq-stream-locator-connection-");
 
     if (clientParametersPrototype.eventLoopGroup == null) {
-      this.eventLoopGroup = new NioEventLoopGroup();
+      this.eventLoopGroup = Utils.eventLoopGroup();
       this.clientParametersPrototype =
           clientParametersPrototype.duplicate().eventLoopGroup(this.eventLoopGroup);
     } else {

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

@@ -20,6 +20,9 @@
 import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import io.netty.channel.ConnectTimeoutException;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
 import java.net.UnknownHostException;
 import java.security.cert.X509Certificate;
 import java.time.Duration;
@@ -406,6 +409,10 @@ static Function<ClientConnectionType, String> defaultConnectionNamingStrategy(St
         prefixes.get(clientConnectionType) + sequences.get(clientConnectionType).getAndIncrement();
   }
 
+  static EventLoopGroup eventLoopGroup() {
+    return new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
+  }
+
   /*
   class to help testing SAC on super streams
    */

src/test/java/com/rabbitmq/stream/DefaultEnvironmentTest.java

@@ -19,7 +19,8 @@
 
 import com.rabbitmq.stream.impl.Client;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
 import java.util.UUID;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
@@ -31,7 +32,7 @@ public class DefaultEnvironmentTest {
 
   @BeforeAll
   static void initAll() {
-    eventLoopGroup = new NioEventLoopGroup();
+    eventLoopGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
   }
 
   @AfterAll

src/test/java/com/rabbitmq/stream/docs/EnvironmentUsage.java

@@ -19,7 +19,9 @@
 import com.rabbitmq.stream.observation.micrometer.MicrometerObservationCollectorBuilder;
 import io.micrometer.observation.ObservationRegistry;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollIoHandler;
 import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
@@ -140,7 +142,9 @@ void deleteStream() {
 
     void nativeEpoll() {
         // tag::native-epoll[]
-        EventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup();  // <1>
+        EventLoopGroup epollEventLoopGroup = new MultiThreadIoEventLoopGroup(  // <1>
+            EpollIoHandler.newFactory()                                        // <1>
+        );                                                                     // <1>
         Environment environment = Environment.builder()
             .netty()  // <2>
                 .eventLoopGroup(epollEventLoopGroup)  // <3>

src/test/java/com/rabbitmq/stream/impl/AlarmsTest.java

@@ -32,7 +32,6 @@
 import com.rabbitmq.stream.Producer;
 import com.rabbitmq.stream.StreamException;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
 import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicReference;
@@ -56,7 +55,7 @@ public class AlarmsTest {
 
   @BeforeAll
   static void initAll() {
-    eventLoopGroup = new NioEventLoopGroup();
+    eventLoopGroup = Utils.eventLoopGroup();
   }
 
   @AfterAll

src/test/java/com/rabbitmq/stream/impl/MqttInteroperabilityTest.java

@@ -27,7 +27,6 @@
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.amqp.UnsignedByte;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
 import java.nio.charset.StandardCharsets;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
@@ -57,7 +56,7 @@ public class MqttInteroperabilityTest {
 
   @BeforeAll
   static void initAll() {
-    eventLoopGroup = new NioEventLoopGroup();
+    eventLoopGroup = Utils.eventLoopGroup();
   }
 
   @AfterAll

src/test/java/com/rabbitmq/stream/impl/StompInteroperabilityTest.java

@@ -22,7 +22,6 @@
 
 import com.rabbitmq.stream.*;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -66,7 +65,7 @@ public class StompInteroperabilityTest {
 
   @BeforeAll
   static void initAll() {
-    eventLoopGroup = new NioEventLoopGroup();
+    eventLoopGroup = Utils.eventLoopGroup();
   }
 
   @AfterAll

src/test/java/com/rabbitmq/stream/impl/StreamEnvironmentTest.java

@@ -51,7 +51,8 @@
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfTlsNotEnabled;
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.epoll.EpollIoHandler;
 import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.handler.ssl.SslHandler;
 import java.net.ConnectException;
@@ -723,7 +724,8 @@ void nettyInitializersAreCalled() {
   @EnabledIfSystemProperty(named = "os.arch", matches = "amd64")
   void nativeEpollWorksOnLinux() {
     int messageCount = 10_000;
-    EventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup();
+    EventLoopGroup epollEventLoopGroup =
+        new MultiThreadIoEventLoopGroup(EpollIoHandler.newFactory());
     try {
       Set<Channel> channels = ConcurrentHashMap.newKeySet();
       try (Environment env =

src/test/java/com/rabbitmq/stream/impl/TestUtils.java

@@ -42,7 +42,6 @@
 import com.rabbitmq.stream.impl.Client.Response;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
 import io.vavr.Tuple2;
 import java.io.IOException;
 import java.lang.annotation.Documented;
@@ -627,7 +626,7 @@ static EventLoopGroup eventLoopGroup(ExtensionContext context) {
 
     @Override
     public void beforeAll(ExtensionContext context) {
-      store(context).put("nettyEventLoopGroup", new NioEventLoopGroup());
+      store(context).put("nettyEventLoopGroup", Utils.eventLoopGroup());
     }
 
     @Override