Update gRPC-Java to 1.30.0 (#2814)

Motivation:

gRPC-Java [1.30.0](https://github.com/grpc/grpc-java/releases/tag/v1.30.0) has been released
We need a workaround for `ProtoReflectionService`. Because it does not implement `InternalNotifyOnServerBuild` anymore.
See grpc/grpc-java#6967 #2806
It is a temporary workaround until upstream handles grpc/grpc-java#7138.

Modifications:

- Upgrade gRPC to 1.30.0 from 1.29.0
- Add ProtoReflectionServiceInterceptor that injects dummy server to
  gRPC context.

Result:

You can run gRPC 1.30.0 with Armeria server.
Fixes #2806

benchmarks/build.gradle

@@ -18,6 +18,7 @@ dependencies {
     implementation project(':thrift')
 
     implementation 'com.squareup.retrofit2:converter-jackson'
+    implementation 'com.google.protobuf:protobuf-java-util'
     implementation 'io.grpc:grpc-okhttp'
     implementation 'io.grpc:grpc-netty-shaded'
     implementation 'org.awaitility:awaitility'

dependencies.yml

@@ -5,7 +5,7 @@
 boms:
   - com.fasterxml.jackson:jackson-bom:2.11.0
   - io.dropwizard.metrics:metrics-bom:4.1.9
-  - io.grpc:grpc-bom:1.29.0
+  - io.grpc:grpc-bom:1.30.0
   - io.micrometer:micrometer-bom:1.5.1
   # NOTE: When changing this, re-evaluate netty-tcnative-boringssl-static below
   - io.netty:netty-bom:4.1.50.Final
@@ -107,7 +107,7 @@ com.google.j2objc:
 # See: https://github.com/grpc/grpc-java/blob/master/build.gradle
 #      (Switch to the right tag and look for 'protobufVersion' and 'protocVersion'.)
 com.google.protobuf:
-  protobuf-java: { version: &PROTOBUF_VERSION '3.11.4' }
+  protobuf-java: { version: &PROTOBUF_VERSION '3.12.0' }
   protobuf-java-util:
     version: *PROTOBUF_VERSION
     exclusions:

examples/grpc-kotlin/gen-src/main/grpc/example/armeria/grpc/kotlin/HelloServiceGrpc.java

@@ -18,7 +18,7 @@
 /**
  */
 @javax.annotation.Generated(
-    value = "by gRPC proto compiler (version 1.29.0)",
+    value = "by gRPC proto compiler (version 1.30.0)",
     comments = "Source: hello.proto")
 public final class HelloServiceGrpc {
 

grpc/src/main/java/com/linecorp/armeria/server/grpc/FramedGrpcService.java

@@ -73,7 +73,6 @@
 import io.grpc.ServerMethodDefinition;
 import io.grpc.ServerServiceDefinition;
 import io.grpc.Status;
-import io.grpc.protobuf.services.ProtoReflectionService;
 
 /**
  * The framed {@link GrpcService} implementation.
@@ -89,13 +88,13 @@ final class FramedGrpcService extends AbstractHttpService implements GrpcService
     private final Set<SerializationFormat> supportedSerializationFormats;
     @Nullable
     private final MessageMarshaller jsonMarshaller;
+    @Nullable
+    private final ProtoReflectionServiceInterceptor protoReflectionServiceInterceptor;
     private final int maxOutboundMessageSizeBytes;
     private final boolean useBlockingTaskExecutor;
     private final boolean unsafeWrapRequestBuffers;
     private final boolean useClientTimeoutHeader;
     private final String advertisedEncodingsHeader;
-    @Nullable
-    private final ProtoReflectionService protoReflectionService;
 
     private final Map<SerializationFormat, ResponseHeaders> defaultHeaders;
 
@@ -107,20 +106,20 @@ final class FramedGrpcService extends AbstractHttpService implements GrpcService
                       CompressorRegistry compressorRegistry,
                       Set<SerializationFormat> supportedSerializationFormats,
                       Consumer<MessageMarshaller.Builder> jsonMarshallerCustomizer,
+                      @Nullable ProtoReflectionServiceInterceptor protoReflectionServiceInterceptor,
                       int maxOutboundMessageSizeBytes,
                       boolean useBlockingTaskExecutor,
                       boolean unsafeWrapRequestBuffers,
                       boolean useClientTimeoutHeader,
-                      @Nullable ProtoReflectionService protoReflectionService,
                       int maxInboundMessageSizeBytes) {
         this.registry = requireNonNull(registry, "registry");
         this.routes = requireNonNull(routes, "routes");
         this.decompressorRegistry = requireNonNull(decompressorRegistry, "decompressorRegistry");
         this.compressorRegistry = requireNonNull(compressorRegistry, "compressorRegistry");
         this.supportedSerializationFormats = supportedSerializationFormats;
         this.useClientTimeoutHeader = useClientTimeoutHeader;
-        this.protoReflectionService = protoReflectionService;
         jsonMarshaller = jsonMarshaller(registry, supportedSerializationFormats, jsonMarshallerCustomizer);
+        this.protoReflectionServiceInterceptor = protoReflectionServiceInterceptor;
         this.maxOutboundMessageSizeBytes = maxOutboundMessageSizeBytes;
         this.useBlockingTaskExecutor = useBlockingTaskExecutor;
         this.unsafeWrapRequestBuffers = unsafeWrapRequestBuffers;
@@ -256,7 +255,7 @@ public void serviceAdded(ServiceConfig cfg) {
             maxInboundMessageSizeBytes = (int) Math.min(cfg.maxRequestLength(), Integer.MAX_VALUE);
         }
 
-        if (protoReflectionService != null) {
+        if (protoReflectionServiceInterceptor != null) {
             final Map<String, ServerServiceDefinition> grpcServices =
                     cfg.server().config().virtualHosts().stream()
                        .flatMap(host -> host.serviceConfigs().stream())
@@ -269,62 +268,66 @@ public void serviceAdded(ServiceConfig cfg) {
                        .collect(toImmutableMap(def -> def.getServiceDescriptor().getName(),
                                                Function.identity(),
                                                (a, b) -> a));
-            protoReflectionService.notifyOnBuild(new Server() {
-                @Override
-                public Server start() {
-                    throw new UnsupportedOperationException();
-                }
+            protoReflectionServiceInterceptor.setServer(newDummyServer(grpcServices));
+        }
+    }
 
-                @Override
-                public List<ServerServiceDefinition> getServices() {
-                    return ImmutableList.copyOf(grpcServices.values());
-                }
+    private static Server newDummyServer(Map<String, ServerServiceDefinition> grpcServices) {
+        return new Server() {
+            @Override
+            public Server start() {
+                throw new UnsupportedOperationException();
+            }
 
-                @Override
-                public List<ServerServiceDefinition> getImmutableServices() {
-                    // NB: This will probably go away in favor of just getServices above, so we
-                    // implement both the same.
-                    // https://github.com/grpc/grpc-java/issues/4600
-                    return getServices();
-                }
+            @Override
+            public List<ServerServiceDefinition> getServices() {
+                return ImmutableList.copyOf(grpcServices.values());
+            }
 
-                @Override
-                public List<ServerServiceDefinition> getMutableServices() {
-                    // Armeria does not have the concept of mutable services.
-                    return ImmutableList.of();
-                }
+            @Override
+            public List<ServerServiceDefinition> getImmutableServices() {
+                // NB: This will probably go away in favor of just getServices above, so we
+                // implement both the same.
+                // https://github.com/grpc/grpc-java/issues/4600
+                return getServices();
+            }
 
-                @Override
-                public Server shutdown() {
-                    throw new UnsupportedOperationException();
-                }
+            @Override
+            public List<ServerServiceDefinition> getMutableServices() {
+                // Armeria does not have the concept of mutable services.
+                return ImmutableList.of();
+            }
 
-                @Override
-                public Server shutdownNow() {
-                    throw new UnsupportedOperationException();
-                }
+            @Override
+            public Server shutdown() {
+                throw new UnsupportedOperationException();
+            }
 
-                @Override
-                public boolean isShutdown() {
-                    throw new UnsupportedOperationException();
-                }
+            @Override
+            public Server shutdownNow() {
+                throw new UnsupportedOperationException();
+            }
 
-                @Override
-                public boolean isTerminated() {
-                    throw new UnsupportedOperationException();
-                }
+            @Override
+            public boolean isShutdown() {
+                throw new UnsupportedOperationException();
+            }
 
-                @Override
-                public boolean awaitTermination(long timeout, TimeUnit unit) {
-                    throw new UnsupportedOperationException();
-                }
+            @Override
+            public boolean isTerminated() {
+                throw new UnsupportedOperationException();
+            }
 
-                @Override
-                public void awaitTermination() {
-                    throw new UnsupportedOperationException();
-                }
-            });
-        }
+            @Override
+            public boolean awaitTermination(long timeout, TimeUnit unit) {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public void awaitTermination() {
+                throw new UnsupportedOperationException();
+            }
+        };
     }
 
     @Override

grpc/src/main/java/com/linecorp/armeria/server/grpc/GrpcServiceBuilder.java

@@ -52,6 +52,7 @@
 import io.grpc.BindableService;
 import io.grpc.CompressorRegistry;
 import io.grpc.DecompressorRegistry;
+import io.grpc.ServerInterceptors;
 import io.grpc.ServerServiceDefinition;
 import io.grpc.protobuf.services.ProtoReflectionService;
 
@@ -71,6 +72,9 @@ public final class GrpcServiceBuilder {
     @Nullable
     private CompressorRegistry compressorRegistry;
 
+    @Nullable
+    private ProtoReflectionServiceInterceptor protoReflectionServiceInterceptor;
+
     private Set<SerializationFormat> supportedSerializationFormats = DEFAULT_SUPPORTED_SERIALIZATION_FORMATS;
 
     private int maxInboundMessageSizeBytes = ArmeriaMessageDeframer.NO_MAX_INBOUND_MESSAGE_SIZE;
@@ -87,9 +91,6 @@ public final class GrpcServiceBuilder {
 
     private boolean useClientTimeoutHeader = true;
 
-    @Nullable
-    private ProtoReflectionService protoReflectionService;
-
     GrpcServiceBuilder() {}
 
     /**
@@ -107,10 +108,11 @@ public GrpcServiceBuilder addService(ServerServiceDefinition service) {
      */
     public GrpcServiceBuilder addService(BindableService bindableService) {
         if (bindableService instanceof ProtoReflectionService) {
-            checkState(protoReflectionService == null,
+            checkState(protoReflectionServiceInterceptor == null,
                        "Attempting to add a ProtoReflectionService but one is already present. " +
                        "ProtoReflectionService must only be added once.");
-            protoReflectionService = (ProtoReflectionService) bindableService;
+            protoReflectionServiceInterceptor = new ProtoReflectionServiceInterceptor();
+            return addService(ServerInterceptors.intercept(bindableService, protoReflectionServiceInterceptor));
         }
 
         return addService(bindableService.bindService());
@@ -308,11 +310,11 @@ public GrpcService build() {
                 firstNonNull(compressorRegistry, CompressorRegistry.getDefaultInstance()),
                 supportedSerializationFormats,
                 jsonMarshallerCustomizer,
+                protoReflectionServiceInterceptor,
                 maxOutboundMessageSizeBytes,
                 useBlockingTaskExecutor,
                 unsafeWrapRequestBuffers,
                 useClientTimeoutHeader,
-                protoReflectionService,
                 maxInboundMessageSizeBytes);
         return enableUnframedRequests ? new UnframedGrpcService(grpcService) : grpcService;
     }

grpc/src/main/java/com/linecorp/armeria/server/grpc/ProtoReflectionServiceInterceptor.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.server.grpc;
+
+import javax.annotation.Nullable;
+
+import io.grpc.Context;
+import io.grpc.Contexts;
+import io.grpc.InternalServer;
+import io.grpc.Metadata;
+import io.grpc.Server;
+import io.grpc.ServerCall;
+import io.grpc.ServerCall.Listener;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+
+final class ProtoReflectionServiceInterceptor implements ServerInterceptor {
+
+    @Nullable
+    private Server server;
+
+    @Override
+    public <I, O> Listener<I> interceptCall(ServerCall<I, O> call, Metadata headers,
+                                            ServerCallHandler<I, O> next) {
+        // Should set server before calling this
+        assert server != null;
+
+        final Context context = Context.current().withValue(InternalServer.SERVER_CONTEXT_KEY, server);
+        return Contexts.interceptCall(context, call, headers, next);
+    }
+
+    public void setServer(Server server) {
+        this.server = server;
+    }
+}

it/server/build.gradle

@@ -2,5 +2,7 @@ dependencies {
     testImplementation project(':grpc')
     testImplementation 'io.grpc:grpc-core'
     testImplementation 'io.grpc:grpc-interop-testing'
+    testImplementation 'io.grpc:grpc-okhttp'
+    testImplementation 'io.grpc:grpc-testing'
     testImplementation 'org.apache.hbase:hbase-shaded-client'
 }