Address PR comments - majorly move away from grpc-netty-shaded

Signed-off-by: Rishabh Maurya <[email protected]>
rishabhmaurya · Feb 10, 2025 · db6d461 · db6d461
1 parent 488572c
commit db6d461
Show file tree

Hide file tree

Showing 33 changed files with 136 additions and 924 deletions.
diff --git a/libs/build.gradle b/libs/build.gradle
@@ -44,12 +44,12 @@ subprojects {
       dependencies.matching { it instanceof ProjectDependency }.all { ProjectDependency dep ->
         Project depProject = project.project(dep.path)
         if (depProject != null
-          && (false == depProject.path.equals(':libs:opensearch-core') && false == depProject.path.equals(':libs:opensearch-arrow-memory-shaded') &&
+          && (false == depProject.path.equals(':libs:opensearch-core') &&
               false == depProject.path.equals(':libs:opensearch-common'))
           && depProject.path.startsWith(':libs')) {
           throw new InvalidUserDataException("projects in :libs "
             + "may not depend on other projects libs except "
-            + ":libs:opensearch-core or :libs:opensearch-common  or :libs:opensearch-arrow-memory-shaded but "
+            + ":libs:opensearch-core or :libs:opensearch-common but "
             + "${project.path} depends on ${depProject.path}")
         }
       }

diff --git a/plugins/arrow-flight-rpc/build.gradle b/plugins/arrow-flight-rpc/build.gradle
@@ -13,7 +13,7 @@ apply plugin: 'opensearch.internal-cluster-test'
 
 opensearchplugin {
   description = 'Arrow flight based Stream implementation'
-  classname = 'org.opensearch.arrow.flight.FlightStreamPlugin'
+  classname = 'org.opensearch.arrow.flight.bootstrap.FlightStreamPlugin'
 }
 
 dependencies {
@@ -27,18 +27,28 @@ dependencies {
 
   runtimeOnly "org.apache.arrow:arrow-memory-netty:${versions.arrow}"
   runtimeOnly "org.apache.arrow:arrow-memory-netty-buffer-patch:${versions.arrow}"
-  runtimeOnly "io.netty:netty-buffer:${versions.netty}"
-  runtimeOnly "io.netty:netty-common:${versions.netty}"
+
+  implementation "io.netty:netty-buffer:${versions.netty}"
+  implementation "io.netty:netty-common:${versions.netty}"
+
+  implementation "io.netty:netty-codec:${versions.netty}"
+  implementation "io.netty:netty-codec-http:${versions.netty}"
+  implementation "io.netty:netty-codec-http2:${versions.netty}"
+  implementation "io.netty:netty-handler:${versions.netty}"
+  implementation "io.netty:netty-resolver:${versions.netty}"
+  implementation "io.netty:netty-transport:${versions.netty}"
+  implementation "io.netty:netty-transport-native-unix-common:${versions.netty}"
+  implementation "io.netty:netty-transport-classes-epoll:${versions.netty}"
+  implementation "io.netty:netty-tcnative-classes:2.0.66.Final"
 
   implementation "org.slf4j:slf4j-api:${versions.slf4j}"
   runtimeOnly "com.google.flatbuffers:flatbuffers-java:${versions.flatbuffers}"
   runtimeOnly "commons-codec:commons-codec:${versions.commonscodec}"
 
-  implementation "io.grpc:grpc-netty-shaded:${versions.grpc}"
-
   implementation "io.grpc:grpc-api:${versions.grpc}"
   runtimeOnly "io.grpc:grpc-core:${versions.grpc}"
   implementation "io.grpc:grpc-stub:${versions.grpc}"
+  implementation "io.grpc:grpc-netty:${versions.grpc}"
 
   runtimeOnly group: 'com.google.code.findbugs', name: 'jsr305', version: '3.0.2'
   compileOnly 'org.immutables:value:2.10.1'
@@ -194,16 +204,7 @@ tasks.named('thirdPartyAudit').configure {
     'reactor.blockhound.BlockHound$Builder',
     'reactor.blockhound.integration.BlockHoundIntegration',
 
-    'com.google.protobuf.util.Timestamps',
-    'io.grpc.netty.GrpcSslContexts',
-    'io.grpc.netty.NettyChannelBuilder',
-    'io.grpc.netty.NettyServerBuilder',
-    'io.netty.channel.EventLoopGroup',
-    'io.netty.channel.ServerChannel',
-    'io.netty.handler.ssl.ClientAuth',
-    'io.netty.handler.ssl.SslContext',
-    'io.netty.handler.ssl.SslContextBuilder',
-    'io.netty.handler.ssl.util.InsecureTrustManagerFactory'
+    'com.google.protobuf.util.Timestamps'
   )
   ignoreViolations(
     // Guava internal classes
@@ -221,32 +222,32 @@ tasks.named('thirdPartyAudit').configure {
     'com.google.common.util.concurrent.AbstractFuture$UnsafeAtomicHelper',
     'com.google.common.util.concurrent.AbstractFuture$UnsafeAtomicHelper$1',
 
-    'io.grpc.netty.shaded.io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator',
-    'io.grpc.netty.shaded.io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$1',
-    'io.grpc.netty.shaded.io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$2',
-    'io.grpc.netty.shaded.io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$3',
-    'io.grpc.netty.shaded.io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$4',
-    'io.grpc.netty.shaded.io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$5',
-    'io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0',
-    'io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0$1',
-    'io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0$2',
-    'io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0$3',
-    'io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0$4',
-    'io.grpc.netty.shaded.io.netty.util.internal.PlatformDependent0$6',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueConsumerNodeRef',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueProducerNodeRef',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueProducerFields',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.LinkedQueueNode',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueConsumerIndexField',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueProducerIndexField',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueConsumerIndexField',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerIndexField',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerLimitField',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.util.UnsafeAccess',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.util.UnsafeLongArrayAccess',
-    'io.grpc.netty.shaded.io.netty.util.internal.shaded.org.jctools.util.UnsafeRefArrayAccess',
+    'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator',
+    'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$1',
+    'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$2',
+    'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$3',
+    'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$4',
+    'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$5',
+    'io.netty.util.internal.PlatformDependent0',
+    'io.netty.util.internal.PlatformDependent0$1',
+    'io.netty.util.internal.PlatformDependent0$2',
+    'io.netty.util.internal.PlatformDependent0$3',
+    'io.netty.util.internal.PlatformDependent0$4',
+    'io.netty.util.internal.PlatformDependent0$6',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueConsumerNodeRef',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueProducerNodeRef',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueProducerFields',
+    'io.netty.util.internal.shaded.org.jctools.queues.LinkedQueueNode',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueConsumerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueProducerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueConsumerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerLimitField',
+    'io.netty.util.internal.shaded.org.jctools.util.UnsafeAccess',
+    'io.netty.util.internal.shaded.org.jctools.util.UnsafeLongArrayAccess',
+    'io.netty.util.internal.shaded.org.jctools.util.UnsafeRefArrayAccess',
     'io.netty.util.internal.PlatformDependent0',
     'io.netty.util.internal.PlatformDependent0$1',
     'io.netty.util.internal.PlatformDependent0$2',

diff --git a/plugins/arrow-flight-rpc/licenses/grpc-netty-1.68.2.jar.sha1 b/plugins/arrow-flight-rpc/licenses/grpc-netty-1.68.2.jar.sha1
@@ -0,0 +1 @@
+3c3279d2e3520195fd26e0c3d9aca2ed1157d8c3
diff --git a/plugins/arrow-flight-rpc/licenses/grpc-netty-shaded-1.68.2.jar.sha1 b/plugins/arrow-flight-rpc/licenses/grpc-netty-shaded-1.68.2.jar.sha1
diff --git a/plugins/arrow-flight-rpc/licenses/netty-codec-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-codec-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+2831d3431ed93d9c0b64b1c0cce2ced4737539aa
diff --git a/plugins/arrow-flight-rpc/licenses/netty-codec-http-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-codec-http-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+5a0f8cd908b8b09b2cd1d39c1d2086a4d12e6029
diff --git a/plugins/arrow-flight-rpc/licenses/netty-codec-http2-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-codec-http2-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+e0a678ac80e00b08a4c0118d496efabc4516ebbf
diff --git a/plugins/arrow-flight-rpc/licenses/netty-handler-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-handler-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+db14cd99515f8c98a3f2a347718e59f14d85c503
diff --git a/plugins/arrow-flight-rpc/licenses/netty-resolver-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-resolver-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+581b37489a03162f473264b65f53d504269a74b0
diff --git a/plugins/arrow-flight-rpc/licenses/netty-tcnative-classes-2.0.66.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-tcnative-classes-2.0.66.Final.jar.sha1
@@ -0,0 +1 @@
+9588bd2f891157538a78d86c945aa34bf9308dda
diff --git a/plugins/arrow-flight-rpc/licenses/netty-transport-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-transport-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+f81d72962bd134d8d8e11b514321134fa5fd0ce6
diff --git a/plugins/arrow-flight-rpc/licenses/netty-transport-classes-epoll-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-transport-classes-epoll-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+0ebca585acd227b8682ed5b2aafbb86d07f77848
diff --git a/plugins/arrow-flight-rpc/licenses/netty-transport-native-unix-common-4.1.117.Final.jar.sha1 b/plugins/arrow-flight-rpc/licenses/netty-transport-native-unix-common-4.1.117.Final.jar.sha1
@@ -0,0 +1 @@
+684f2316ff2b2171babbc17c95ac3bd97f5f091e
diff --git a/...ght-rpc/src/internalClusterTest/java/org/opensearch/arrow/flight/ArrowFlightServerIT.java b/...ght-rpc/src/internalClusterTest/java/org/opensearch/arrow/flight/ArrowFlightServerIT.java
@@ -12,6 +12,7 @@
 import org.apache.arrow.flight.OSFlightClient;
 import org.opensearch.arrow.flight.bootstrap.FlightClientManager;
 import org.opensearch.arrow.flight.bootstrap.FlightService;
+import org.opensearch.arrow.flight.bootstrap.FlightStreamPlugin;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.plugins.Plugin;

diff --git a/plugins/arrow-flight-rpc/src/main/java/org/apache/arrow/flight/OSFlightClient.java b/plugins/arrow-flight-rpc/src/main/java/org/apache/arrow/flight/OSFlightClient.java
@@ -54,17 +54,17 @@
 import io.grpc.ManagedChannel;
 import io.grpc.MethodDescriptor;
 import io.grpc.StatusRuntimeException;
-import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
-import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
-import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup;
-import io.grpc.netty.shaded.io.netty.channel.ServerChannel;
-import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
-import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
-import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import io.grpc.netty.GrpcSslContexts;
+import io.grpc.netty.NettyChannelBuilder;
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.ClientCalls;
 import io.grpc.stub.ClientResponseObserver;
 import io.grpc.stub.StreamObserver;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.ServerChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 
 /**
  * Clone of {@link FlightClient} to support setting SslContext directly. It can be discarded once
@@ -718,7 +718,7 @@ public static Builder builder(BufferAllocator allocator, Location location) {
     public static Builder builder(
         BufferAllocator allocator,
         Location location,
-        Class<? extends io.grpc.netty.shaded.io.netty.channel.Channel> channelType,
+        Class<? extends io.netty.channel.Channel> channelType,
         ExecutorService executorService,
         EventLoopGroup workerELG,
         SslContext sslContext
@@ -747,7 +747,7 @@ public static final class Builder {
         private boolean verifyServer = true;
         private EventLoopGroup workerELG;
         private ExecutorService executorService;
-        private Class<? extends io.grpc.netty.shaded.io.netty.channel.Channel> channelType;
+        private Class<? extends io.netty.channel.Channel> channelType;
         private SslContext sslContext;
 
         private Builder() {}
@@ -820,7 +820,7 @@ public Builder executor(ExecutorService executorService) {
             return this;
         }
 
-        public Builder channelType(Class<? extends io.grpc.netty.shaded.io.netty.channel.Channel> channelType) {
+        public Builder channelType(Class<? extends io.netty.channel.Channel> channelType) {
             this.channelType = channelType;
             return this;
         }
@@ -860,10 +860,9 @@ public OSFlightClient build() {
                         try {
                             // Linux
                             builder.channelType(
-                                Class.forName("io.grpc.netty.shaded.io.netty.channel.epoll.EpollDomainSocketChannel")
-                                    .asSubclass(ServerChannel.class)
+                                Class.forName("io.netty.channel.epoll.EpollDomainSocketChannel").asSubclass(ServerChannel.class)
                             );
-                            final EventLoopGroup elg = Class.forName("io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup")
+                            final EventLoopGroup elg = Class.forName("io.netty.channel.epoll.EpollEventLoopGroup")
                                 .asSubclass(EventLoopGroup.class)
                                 .getDeclaredConstructor()
                                 .newInstance();
@@ -872,10 +871,9 @@ public OSFlightClient build() {
                             // BSD
                             // this might not work as io.netty.channel.kqueue classes aren't present in grpc-netty-shaded
                             builder.channelType(
-                                Class.forName("io.grpc.netty.shaded.io.netty.channel.kqueue.KQueueDomainSocketChannel")
-                                    .asSubclass(ServerChannel.class)
+                                Class.forName("io.netty.channel.kqueue.KQueueDomainSocketChannel").asSubclass(ServerChannel.class)
                             );
-                            final EventLoopGroup elg = Class.forName("io.grpc.netty.shaded.io.netty.channel.kqueue.KQueueEventLoopGroup")
+                            final EventLoopGroup elg = Class.forName("io.netty.channel.kqueue.KQueueEventLoopGroup")
                                 .asSubclass(EventLoopGroup.class)
                                 .getDeclaredConstructor()
                                 .newInstance();

diff --git a/plugins/arrow-flight-rpc/src/main/java/org/apache/arrow/flight/OSFlightServer.java b/plugins/arrow-flight-rpc/src/main/java/org/apache/arrow/flight/OSFlightServer.java
@@ -44,14 +44,14 @@
 
 import io.grpc.Server;
 import io.grpc.ServerInterceptors;
-import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
-import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
-import io.grpc.netty.shaded.io.netty.channel.Channel;
-import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup;
-import io.grpc.netty.shaded.io.netty.channel.ServerChannel;
-import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
-import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
-import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
+import io.grpc.netty.GrpcSslContexts;
+import io.grpc.netty.NettyServerBuilder;
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.ServerChannel;
+import io.netty.handler.ssl.ClientAuth;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
 
 /**
  * Clone of {@link FlightServer} to support setting SslContext directly. It can be discarded once
@@ -243,10 +243,9 @@ public OSFlightServer build() {
                         try {
                             // Linux
                             builder.channelType(
-                                Class.forName("io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerDomainSocketChannel")
-                                    .asSubclass(ServerChannel.class)
+                                Class.forName("io.netty.channel.epoll.EpollServerDomainSocketChannel").asSubclass(ServerChannel.class)
                             );
-                            final EventLoopGroup elg = Class.forName("io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup")
+                            final EventLoopGroup elg = Class.forName("io.netty.channel.epoll.EpollEventLoopGroup")
                                 .asSubclass(EventLoopGroup.class)
                                 .getConstructor()
                                 .newInstance();
@@ -255,10 +254,9 @@ public OSFlightServer build() {
                             // BSD
                             // below logic may not work as the kqueue classes aren't present in grpc-netty-shaded
                             builder.channelType(
-                                Class.forName("io.grpc.netty.shaded.io.netty.channel.kqueue.KQueueServerDomainSocketChannel")
-                                    .asSubclass(ServerChannel.class)
+                                Class.forName("io.netty.channel.kqueue.KQueueServerDomainSocketChannel").asSubclass(ServerChannel.class)
                             );
-                            final EventLoopGroup elg = Class.forName("io.grpc.netty.shaded.io.netty.channel.kqueue.KQueueEventLoopGroup")
+                            final EventLoopGroup elg = Class.forName("io.netty.channel.kqueue.KQueueEventLoopGroup")
                                 .asSubclass(EventLoopGroup.class)
                                 .getConstructor()
                                 .newInstance();