ARROW-10205: [Java][FlightRPC] Allow disabling server validation

java/flight/flight-core/src/main/java/org/apache/arrow/flight/FlightClient.java

@@ -62,6 +62,7 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.ServerChannel;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 
 /**
  * Client for Flight services.
@@ -533,6 +534,7 @@ public static final class Builder {
     private InputStream clientKey = null;
     private String overrideHostname = null;
     private List<FlightClientMiddleware.Factory> middleware = new ArrayList<>();
+    private boolean verifyServer = true;
 
     private Builder() {
     }
@@ -592,6 +594,11 @@ public Builder intercept(FlightClientMiddleware.Factory factory) {
       return this;
     }
 
+    public Builder verifyServer(boolean verifyServer) {
+      this.verifyServer = verifyServer;
+      return this;
+    }
+
     /**
      * Create the client from this builder.
      */
@@ -637,19 +644,29 @@ public FlightClient build() {
       if (this.forceTls || LocationSchemes.GRPC_TLS.equals(location.getUri().getScheme())) {
         builder.useTransportSecurity();
 
-        if (this.trustedCertificates != null || this.clientCertificate != null || this.clientKey != null) {
-          final SslContextBuilder sslContextBuilder = GrpcSslContexts.forClient();
+        final boolean hasTrustedCerts = this.trustedCertificates != null;
+        final boolean hasKeyCertPair = this.clientCertificate != null && this.clientKey != null;
+        if (!this.verifyServer && (hasTrustedCerts || hasKeyCertPair)) {
+          throw new IllegalArgumentException("FlightClient has been configured to disable server verification, " +
+              "but certificate options have been specified.");
+        }
+
+        final SslContextBuilder sslContextBuilder = GrpcSslContexts.forClient();
+
+        if (!this.verifyServer) {
+          sslContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);
+        } else if (this.trustedCertificates != null || this.clientCertificate != null || this.clientKey != null) {
           if (this.trustedCertificates != null) {
             sslContextBuilder.trustManager(this.trustedCertificates);
           }
           if (this.clientCertificate != null && this.clientKey != null) {
             sslContextBuilder.keyManager(this.clientCertificate, this.clientKey);
           }
-          try {
-            builder.sslContext(sslContextBuilder.build());
-          } catch (SSLException e) {
-            throw new RuntimeException(e);
-          }
+        }
+        try {
+          builder.sslContext(sslContextBuilder.build());
+        } catch (SSLException e) {
+          throw new RuntimeException(e);
         }
 
         if (this.overrideHostname != null) {

java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestTls.java

@@ -85,6 +85,22 @@ public void rejectHostname() {
     });
   }
 
+  /**
+   * Test a basic request over TLS.
+   */
+  @Test
+  public void connectTlsDisableServerVerification() {
+    test((builder) -> {
+      try (final FlightClient client = builder.verifyServer(false).build()) {
+        final Iterator<Result> responses = client.doAction(new Action("hello-world"));
+        final byte[] response = responses.next().getBody();
+        Assert.assertEquals("Hello, world!", new String(response, StandardCharsets.UTF_8));
+        Assert.assertFalse(responses.hasNext());
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    });
+  }
 
   void test(Consumer<Builder> testFn) {
     final FlightTestUtil.CertKeyPair certKey = FlightTestUtil.exampleTlsCerts().get(0);