ARROW-7477: [Java][FlightRPC] set up gRPC reflection metadata

This sets up the gRPC/Protobuf reflection metadata, so that [gRPC reflection](https://github.com/grpc/grpc/blob/master/doc/server-reflection.md) works. This is useful for debugging. Additionally, it means interceptors can do generic introspection on gRPC and Flight methods, without having to special-case them. (You might want to do this to implement path-based access control, for instance.)

Closes #6114 from lidavidm/arrow-7477-reflection and squashes the following commits:

8b763dd <David Li> ARROW-7477:  set up gRPC reflection metadata

Authored-by: David Li <[email protected]>
Signed-off-by: Wes McKinney <[email protected]>

Author: lidavidm

java/flight/src/main/java/org/apache/arrow/flight/FlightBindingService.java

@@ -23,6 +23,7 @@
 import org.apache.arrow.flight.auth.ServerAuthHandler;
 import org.apache.arrow.flight.impl.Flight;
 import org.apache.arrow.flight.impl.Flight.PutResult;
+import org.apache.arrow.flight.impl.FlightServiceGrpc;
 import org.apache.arrow.memory.BufferAllocator;
 
 import com.google.common.collect.ImmutableSet;
@@ -32,6 +33,7 @@
 import io.grpc.MethodDescriptor.MethodType;
 import io.grpc.ServerMethodDefinition;
 import io.grpc.ServerServiceDefinition;
+import io.grpc.ServiceDescriptor;
 import io.grpc.protobuf.ProtoUtils;
 import io.grpc.stub.ServerCalls;
 import io.grpc.stub.StreamObserver;
@@ -61,6 +63,7 @@ public static MethodDescriptor<Flight.Ticket, ArrowMessage> getDoGetDescriptor(B
         .setSampledToLocalTracing(false)
         .setRequestMarshaller(ProtoUtils.marshaller(Flight.Ticket.getDefaultInstance()))
         .setResponseMarshaller(ArrowMessage.createMarshaller(allocator))
+        .setSchemaDescriptor(FlightServiceGrpc.getDoGetMethod().getSchemaDescriptor())
         .build();
   }
 
@@ -71,6 +74,7 @@ public static MethodDescriptor<ArrowMessage, Flight.PutResult> getDoPutDescripto
         .setSampledToLocalTracing(false)
         .setRequestMarshaller(ArrowMessage.createMarshaller(allocator))
         .setResponseMarshaller(ProtoUtils.marshaller(Flight.PutResult.getDefaultInstance()))
+        .setSchemaDescriptor(FlightServiceGrpc.getDoPutMethod().getSchemaDescriptor())
         .build();
   }
 
@@ -82,7 +86,21 @@ public ServerServiceDefinition bindService() {
 
     final MethodDescriptor<ArrowMessage, Flight.PutResult> doPutDescriptor = getDoPutDescriptor(allocator);
 
-    ServerServiceDefinition.Builder serviceBuilder = ServerServiceDefinition.builder(FlightConstants.SERVICE);
+    // Make sure we preserve SchemaDescriptor fields on methods so that gRPC reflection still works.
+    final ServiceDescriptor.Builder serviceDescriptorBuilder = ServiceDescriptor.newBuilder(FlightConstants.SERVICE)
+        .setSchemaDescriptor(baseDefinition.getServiceDescriptor().getSchemaDescriptor());
+    serviceDescriptorBuilder.addMethod(doGetDescriptor);
+    serviceDescriptorBuilder.addMethod(doPutDescriptor);
+    for (MethodDescriptor<?, ?> definition : baseDefinition.getServiceDescriptor().getMethods()) {
+      if (OVERRIDE_METHODS.contains(definition.getFullMethodName())) {
+        continue;
+      }
+
+      serviceDescriptorBuilder.addMethod(definition);
+    }
+
+    final ServiceDescriptor serviceDescriptor = serviceDescriptorBuilder.build();
+    ServerServiceDefinition.Builder serviceBuilder = ServerServiceDefinition.builder(serviceDescriptor);
     serviceBuilder.addMethod(doGetDescriptor, ServerCalls.asyncServerStreamingCall(new DoGetMethod(delegate)));
     serviceBuilder.addMethod(doPutDescriptor, ServerCalls.asyncBidiStreamingCall(new DoPutMethod(delegate)));
 

java/flight/src/test/java/org/apache/arrow/flight/TestServerOptions.java

@@ -17,13 +17,20 @@
 
 package org.apache.arrow.flight;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
 import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 
 import org.apache.arrow.flight.TestBasicOperation.Producer;
+import org.apache.arrow.flight.auth.ServerAuthHandler;
+import org.apache.arrow.flight.impl.FlightServiceGrpc;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.vector.IntVector;
@@ -34,6 +41,8 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
+import io.grpc.MethodDescriptor;
+import io.grpc.ServerServiceDefinition;
 import io.grpc.netty.NettyServerBuilder;
 
 @RunWith(JUnit4.class)
@@ -69,7 +78,7 @@ public void defaultExecutorClosed() throws Exception {
                 (location) -> FlightServer.builder(a, location, new NoOpFlightProducer())
                     .build()
             )) {
-      Assert.assertNotNull(server.grpcExecutor);
+      assertNotNull(server.grpcExecutor);
       executor = server.grpcExecutor;
     }
     Assert.assertTrue(executor.isShutdown());
@@ -128,4 +137,39 @@ public void domainSocket() throws Exception {
       }
     }
   }
+
+  @Test
+  public void checkReflectionMetadata() {
+    // This metadata is needed for gRPC reflection to work.
+    final ExecutorService executorService = Executors.newSingleThreadExecutor();
+    try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE)) {
+      final FlightBindingService service = new FlightBindingService(allocator, new NoOpFlightProducer(),
+          ServerAuthHandler.NO_OP, executorService);
+      final ServerServiceDefinition definition = service.bindService();
+      assertEquals(FlightServiceGrpc.getServiceDescriptor().getSchemaDescriptor(),
+          definition.getServiceDescriptor().getSchemaDescriptor());
+
+      final Map<String, MethodDescriptor<?, ?>> definedMethods = new HashMap<>();
+      final Map<String, MethodDescriptor<?, ?>> serviceMethods = new HashMap<>();
+
+      // Make sure that the reflection metadata object is identical across all the places where it's accessible
+      definition.getMethods().forEach(
+          method -> definedMethods.put(method.getMethodDescriptor().getFullMethodName(), method.getMethodDescriptor()));
+      definition.getServiceDescriptor().getMethods().forEach(
+          method -> serviceMethods.put(method.getFullMethodName(), method));
+
+      for (final MethodDescriptor<?, ?> descriptor : FlightServiceGrpc.getServiceDescriptor().getMethods()) {
+        final String methodName = descriptor.getFullMethodName();
+        Assert.assertTrue("Method is missing from ServerServiceDefinition: " + methodName,
+            definedMethods.containsKey(methodName));
+        Assert.assertTrue("Method is missing from ServiceDescriptor: " + methodName,
+            definedMethods.containsKey(methodName));
+
+        assertEquals(descriptor.getSchemaDescriptor(), definedMethods.get(methodName).getSchemaDescriptor());
+        assertEquals(descriptor.getSchemaDescriptor(), serviceMethods.get(methodName).getSchemaDescriptor());
+      }
+    } finally {
+      executorService.shutdown();
+    }
+  }
 }