Allow InstantiatingGrpcChannelProvider to specify maxHeaderListSize (#581)

Allow InstantiatingGrpcChannelProvider to specify maxHeaderListSize.
When the value is specified, use NettyChannelBuilder to create the
channel.

Author: yihanzhen

build.gradle

@@ -112,7 +112,8 @@ subprojects {
         grpcStub: "io.grpc:grpc-stub:${grpcVersion}",
         grpcAuth: "io.grpc:grpc-auth:${grpcVersion}",
         grpcProtobuf: "io.grpc:grpc-protobuf:${grpcVersion}",
-        grpcNetty: "io.grpc:grpc-netty-shaded:${grpcVersion}",
+        grpcNetty: "io.grpc:grpc-netty:${grpcVersion}",
+        grpcNettyShaded: "io.grpc:grpc-netty-shaded:${grpcVersion}",
         gson: "com.google.code.gson:gson:2.7",
         guava: 'com.google.guava:guava:20.0',
         jsr305: 'com.google.code.findbugs:jsr305:3.0.2',

gax-grpc/build.gradle

@@ -12,9 +12,11 @@ dependencies {
     libraries.auth,
     libraries.authCredentials,
     libraries.commonProtos,
-    libraries.apiCommon
+    libraries.apiCommon,
+    libraries.grpcNettyShaded
 
   compileOnly libraries.autovalue
+  compileOnly libraries.grpcNetty
 
   testCompile project(':gax').sourceSets.test.output,
     libraries.junit,

gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java

@@ -68,6 +68,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   private final String endpoint;
   @Nullable private final GrpcInterceptorProvider interceptorProvider;
   @Nullable private final Integer maxInboundMessageSize;
+  @Nullable private final Integer maxInboundMetadataSize;
   @Nullable private final Duration keepAliveTime;
   @Nullable private final Duration keepAliveTimeout;
   @Nullable private final Boolean keepAliveWithoutCalls;
@@ -80,6 +81,7 @@ private InstantiatingGrpcChannelProvider(Builder builder) {
     this.endpoint = builder.endpoint;
     this.interceptorProvider = builder.interceptorProvider;
     this.maxInboundMessageSize = builder.maxInboundMessageSize;
+    this.maxInboundMetadataSize = builder.maxInboundMetadataSize;
     this.keepAliveTime = builder.keepAliveTime;
     this.keepAliveTimeout = builder.keepAliveTimeout;
     this.keepAliveWithoutCalls = builder.keepAliveWithoutCalls;
@@ -181,12 +183,38 @@ private ManagedChannel createSingleChannel() throws IOException {
     int port = Integer.parseInt(endpoint.substring(colon + 1));
     String serviceAddress = endpoint.substring(0, colon);
 
-    ManagedChannelBuilder builder =
-        ManagedChannelBuilder.forAddress(serviceAddress, port)
-            .intercept(headerInterceptor)
-            .intercept(metadataHandlerInterceptor)
-            .userAgent(headerInterceptor.getUserAgentHeader())
-            .executor(executor);
+    // TODO(hzyi): Use NettyChannelBuilder when maxInboundMetadataSize is specified to unblock Spanner.
+    // Change to ManagedChannelBuilder when https://github.com/grpc/grpc-java/issues/4050 is
+    // resolved.
+    ManagedChannelBuilder builder;
+    if (maxInboundMetadataSize != null) {
+      try {
+        Class.forName("io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder");
+        builder =
+            io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder.forAddress(serviceAddress, port)
+                .maxHeaderListSize(maxInboundMetadataSize);
+      } catch (ClassNotFoundException e) {
+        try {
+          Class.forName("io.grpc.netty.NettyChannelBuilder");
+          builder =
+              io.grpc.netty.NettyChannelBuilder.forAddress(serviceAddress, port)
+                  .maxHeaderListSize(maxInboundMetadataSize);
+        } catch (ClassNotFoundException ex) {
+          throw new RuntimeException(
+              "Unable to create the channel because neither"
+                  + " \"io.grpc.netty.NettyChannelBuilder\" nor"
+                  + " \"io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder\" is found.");
+        }
+      }
+    } else {
+      builder = ManagedChannelBuilder.forAddress(serviceAddress, port);
+    }
+
+    builder
+        .intercept(headerInterceptor)
+        .intercept(metadataHandlerInterceptor)
+        .userAgent(headerInterceptor.getUserAgentHeader())
+        .executor(executor);
     if (maxInboundMessageSize != null) {
       builder.maxInboundMessageSize(maxInboundMessageSize);
     }
@@ -226,6 +254,12 @@ public Boolean getKeepAliveWithoutCalls() {
     return keepAliveWithoutCalls;
   }
 
+  /** The maximum metadata size allowed to be received on the channel. */
+  @BetaApi("The surface for maximum metadata size is not stable yet and may change in the future.")
+  public Integer getMaxInboundMetadataSize() {
+    return maxInboundMetadataSize;
+  }
+
   @Override
   public boolean shouldAutoClose() {
     return true;
@@ -246,6 +280,7 @@ public static final class Builder {
     private String endpoint;
     @Nullable private GrpcInterceptorProvider interceptorProvider;
     @Nullable private Integer maxInboundMessageSize;
+    @Nullable private Integer maxInboundMetadataSize;
     @Nullable private Duration keepAliveTime;
     @Nullable private Duration keepAliveTimeout;
     @Nullable private Boolean keepAliveWithoutCalls;
@@ -262,6 +297,7 @@ private Builder(InstantiatingGrpcChannelProvider provider) {
       this.endpoint = provider.endpoint;
       this.interceptorProvider = provider.interceptorProvider;
       this.maxInboundMessageSize = provider.maxInboundMessageSize;
+      this.maxInboundMetadataSize = provider.maxInboundMetadataSize;
       this.keepAliveTime = provider.keepAliveTime;
       this.keepAliveTimeout = provider.keepAliveTimeout;
       this.keepAliveWithoutCalls = provider.keepAliveWithoutCalls;
@@ -333,6 +369,21 @@ public Integer getMaxInboundMessageSize() {
       return maxInboundMessageSize;
     }
 
+    /** The maximum metadata size allowed to be received on the channel. */
+    @BetaApi(
+        "The surface for maximum metadata size is not stable yet and may change in the future.")
+    public Builder setMaxInboundMetadataSize(Integer max) {
+      this.maxInboundMetadataSize = max;
+      return this;
+    }
+
+    /** The maximum metadata size allowed to be received on the channel. */
+    @BetaApi(
+        "The surface for maximum metadata size is not stable yet and may change in the future.")
+    public Integer getMaxInboundMetadataSize() {
+      return maxInboundMetadataSize;
+    }
+
     /** The time without read activity before sending a keepalive ping. */
     public Builder setKeepAliveTime(Duration duration) {
       this.keepAliveTime = duration;

gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java

@@ -90,6 +90,13 @@ public void testKeepAlive() {
     assertEquals(provider.getKeepAliveWithoutCalls(), keepaliveWithoutCalls);
   }
 
+  @Test
+  public void testMaxInboundMetadataSize() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().setMaxInboundMetadataSize(4096).build();
+    assertThat(provider.getMaxInboundMetadataSize()).isEqualTo(4096);
+  }
+
   @Test
   public void testCpuPoolSize() {
     // happy path