Fix Netty deprecation warnings in transport-reactor-netty4 module

CHANGELOG.md

@@ -34,6 +34,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Allow removing plugin that's optionally extended ([#20417](https://github.com/opensearch-project/OpenSearch/pull/20417))
 - Fix indexing regression and bug fixes for grouping criteria. ([20145](https://github.com/opensearch-project/OpenSearch/pull/20145))
 - LeafReader should not remove SubReaderWrappers incase IndexWriter encounters a non aborting Exception ([#20193](https://github.com/opensearch-project/OpenSearch/pull/20193))
+- Fix Netty deprecation warnings in transport-reactor-netty4 module ([20429](https://github.com/opensearch-project/OpenSearch/pull/20429))
 
 ### Dependencies
 - Bump `com.google.auth:google-auth-library-oauth2-http` from 1.38.0 to 1.41.0 ([#20183](https://github.com/opensearch-project/OpenSearch/pull/20183))

plugins/transport-reactor-netty4/src/main/java/org/opensearch/transport/reactor/SharedGroupFactory.java

@@ -26,7 +26,8 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
 import io.netty.util.concurrent.Future;
 
 import static org.opensearch.common.util.concurrent.OpenSearchExecutors.daemonThreadFactory;
@@ -87,9 +88,10 @@ public synchronized SharedGroup getHttpGroup() {
             return getGenericGroup();
         } else {
             if (dedicatedHttpGroup == null) {
-                NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(
+                EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(
                     httpWorkerCount,
-                    daemonThreadFactory(settings, HttpServerTransport.HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)
+                    daemonThreadFactory(settings, HttpServerTransport.HTTP_SERVER_WORKER_THREAD_NAME_PREFIX),
+                    NioIoHandler.newFactory()
                 );
                 dedicatedHttpGroup = new SharedGroup(new RefCountedGroup(eventLoopGroup));
             }
@@ -99,9 +101,10 @@ public synchronized SharedGroup getHttpGroup() {
 
     private SharedGroup getGenericGroup() {
         if (genericGroup == null) {
-            EventLoopGroup eventLoopGroup = new NioEventLoopGroup(
+            EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(
                 workerCount,
-                daemonThreadFactory(settings, TcpTransport.TRANSPORT_WORKER_THREAD_NAME_PREFIX)
+                daemonThreadFactory(settings, TcpTransport.TRANSPORT_WORKER_THREAD_NAME_PREFIX),
+                NioIoHandler.newFactory()
             );
             this.genericGroup = new RefCountedGroup(eventLoopGroup);
         } else {

plugins/transport-reactor-netty4/src/test/java/org/opensearch/http/reactor/netty4/ReactorHttpClient.java

@@ -36,7 +36,9 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
 import io.netty.handler.codec.http.DefaultFullHttpRequest;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
 import io.netty.handler.codec.http.EmptyHttpHeaders;
@@ -199,15 +201,15 @@ private List<FullHttpResponse> sendRequests(
         final Collection<FullHttpRequest> requests,
         boolean ordered
     ) {
-        final NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
+        final EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
         try {
             final HttpClient client = createClient(remoteAddress, eventLoopGroup);
 
             @SuppressWarnings("unchecked")
             final Mono<FullHttpResponse>[] monos = requests.stream()
                 .map(
                     request -> client.headers(h -> h.add(request.headers()))
-                        .baseUrl(request.getUri())
+                        .baseUrl(request.uri())
                         .request(request.method())
                         .send(Mono.fromSupplier(() -> request.content()))
                         .responseSingle(
@@ -240,12 +242,12 @@ private FullHttpResponse sendRequestStream(
         final HttpRequest request,
         final Stream<ToXContent> stream
     ) {
-        final NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
+        final EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
         try {
             final HttpClient client = createClient(remoteAddress, eventLoopGroup);
 
             return client.headers(h -> h.add(request.headers()))
-                .baseUrl(request.getUri())
+                .baseUrl(request.uri())
                 .request(request.method())
                 .send(Flux.fromStream(stream).map(s -> {
                     try (XContentBuilder builder = XContentType.JSON.contentBuilder()) {
@@ -276,7 +278,7 @@ private FullHttpResponse sendRequestStream(
         }
     }
 
-    private HttpClient createClient(final InetSocketAddress remoteAddress, final NioEventLoopGroup eventLoopGroup) {
+    private HttpClient createClient(final InetSocketAddress remoteAddress, final EventLoopGroup eventLoopGroup) {
         final HttpClient client = HttpClient.newConnection()
             .resolver(DefaultAddressResolverGroup.INSTANCE)
             .runOn(eventLoopGroup)

plugins/transport-reactor-netty4/src/test/java/org/opensearch/http/reactor/netty4/ReactorNetty4HttpServerTransportTests.java

@@ -75,7 +75,9 @@
 import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.DefaultFullHttpRequest;
@@ -504,7 +506,7 @@ public void testConnectTimeout() throws Exception {
             new TimeValue(randomIntBetween(100, 300))
         ).build();
 
-        NioEventLoopGroup group = new NioEventLoopGroup();
+        EventLoopGroup group = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
         try (
             ReactorNetty4HttpServerTransport transport = new ReactorNetty4HttpServerTransport(
                 settings,

plugins/transport-reactor-netty4/src/test/java/org/opensearch/http/reactor/netty4/ssl/SecureReactorNetty4HttpServerTransportTests.java

@@ -67,7 +67,9 @@
 import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.DefaultFullHttpRequest;
@@ -511,7 +513,7 @@ public void testConnectTimeout() throws Exception {
             new TimeValue(randomIntBetween(100, 300))
         ).build();
 
-        NioEventLoopGroup group = new NioEventLoopGroup();
+        EventLoopGroup group = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
         try (
             ReactorNetty4HttpServerTransport transport = new ReactorNetty4HttpServerTransport(
                 settings,