feat: set grpc options

Author: NguyenHoangSon96

src/main/java/com/influxdb/v3/client/config/ClientConfig.java

@@ -61,7 +61,6 @@
  *     <li><code>authenticator</code> - HTTP proxy authenticator</li>
  *     <li><code>headers</code> - headers to be added to requests</li>
  *     <li><code>sslRootsFilePath</code> - path to the stored certificates file in PEM format</li>
- *      <li><code>maxInboundMessageSize</code> - RPC maximum inbound message size that the client can receive</li>
  * </ul>
  * <p>
  * If you want to create a client with custom configuration, you can use following code:
@@ -103,7 +102,6 @@ public final class ClientConfig {
     private final Authenticator authenticator;
     private final Map<String, String> headers;
     private final String sslRootsFilePath;
-    private final Integer maxInboundMessageSize;
 
     /**
      * Deprecated use {@link #proxyUrl}.
@@ -242,16 +240,6 @@ public String getProxyUrl() {
         return proxyUrl;
     }
 
-    /**
-     * Gets rpc max message size client can receive.
-     *
-     * @return the size in Integer, may be null
-     */
-    @Nullable
-    public Integer getMaxInboundMessageSize() {
-        return maxInboundMessageSize;
-    }
-
     /**
      * Gets certificates file path.
      *
@@ -315,8 +303,7 @@ public boolean equals(final Object o) {
                 && Objects.equals(proxyUrl, that.proxyUrl)
                 && Objects.equals(authenticator, that.authenticator)
                 && Objects.equals(headers, that.headers)
-                && Objects.equals(sslRootsFilePath, that.sslRootsFilePath)
-                && Objects.equals(maxInboundMessageSize, that.maxInboundMessageSize);
+                && Objects.equals(sslRootsFilePath, that.sslRootsFilePath);
     }
 
     @Override
@@ -325,7 +312,7 @@ public int hashCode() {
                 database, writePrecision, gzipThreshold,
                 timeout, allowHttpRedirects, disableServerCertificateValidation,
                 proxy, proxyUrl, authenticator, headers,
-                defaultTags, sslRootsFilePath, maxInboundMessageSize);
+                defaultTags, sslRootsFilePath);
     }
 
     @Override
@@ -345,7 +332,6 @@ public String toString() {
                 .add("headers=" + headers)
                 .add("defaultTags=" + defaultTags)
                 .add("sslRootsFilePath=" + sslRootsFilePath)
-                .add("maxInboundMessageSize=" + maxInboundMessageSize)
                 .toString();
     }
 
@@ -371,7 +357,6 @@ public static final class Builder {
         private Authenticator authenticator;
         private Map<String, String> headers;
         private String sslRootsFilePath;
-        private Integer maxInboundMessageSize;
 
         /**
          * Sets the URL of the InfluxDB server.
@@ -604,19 +589,6 @@ public Builder sslRootsFilePath(@Nullable final String sslRootsFilePath) {
             return this;
         }
 
-        /**
-         * Set rpc max message size client can receive. Default is 'null'.
-         *
-         * @param maxInboundMessageSize The size in Integer
-         * @return this
-         */
-        @Nonnull
-        public Builder maxInboundMessageSize(@Nullable final Integer maxInboundMessageSize) {
-
-            this.maxInboundMessageSize = maxInboundMessageSize;
-            return this;
-        }
-
         /**
          * Build an instance of {@code ClientConfig}.
          *
@@ -756,6 +728,5 @@ private ClientConfig(@Nonnull final Builder builder) {
         authenticator = builder.authenticator;
         headers = builder.headers;
         sslRootsFilePath = builder.sslRootsFilePath;
-        maxInboundMessageSize = builder.maxInboundMessageSize;
     }
 }

src/main/java/com/influxdb/v3/client/internal/FlightSqlClient.java

@@ -27,11 +27,13 @@
 import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Objects;
 import java.util.Spliterator;
 import java.util.Spliterators;
 import java.util.stream.Stream;
@@ -49,6 +51,7 @@
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import org.apache.arrow.flight.CallOption;
 import org.apache.arrow.flight.FlightClient;
 import org.apache.arrow.flight.FlightGrpcUtils;
 import org.apache.arrow.flight.FlightStream;
@@ -106,7 +109,8 @@ Stream<VectorSchemaRoot> execute(@Nonnull final String query,
                                      @Nonnull final String database,
                                      @Nonnull final QueryType queryType,
                                      @Nonnull final Map<String, Object> queryParameters,
-                                     @Nonnull final Map<String, String> headers) {
+                                     @Nonnull final Map<String, String> headers,
+                                     final CallOption... callOption) {
 
         Map<String, Object> ticketData = new HashMap<>() {{
             put("database", database);
@@ -126,8 +130,10 @@ Stream<VectorSchemaRoot> execute(@Nonnull final String query,
         }
 
         HeaderCallOption headerCallOption = metadataHeader(headers);
+        CallOption[] callOptions = concatCallOptions(callOption, headerCallOption);
+
         Ticket ticket = new Ticket(json.getBytes(StandardCharsets.UTF_8));
-        FlightStream stream = client.getStream(ticket, headerCallOption);
+        FlightStream stream = client.getStream(ticket, callOptions);
         FlightSqlIterator iterator = new FlightSqlIterator(stream);
 
         Spliterator<VectorSchemaRoot> spliterator = Spliterators.spliteratorUnknownSize(iterator, Spliterator.NONNULL);
@@ -162,13 +168,8 @@ private FlightClient createFlightClient(@Nonnull final ClientConfig config) {
             LOG.warn("proxy property in ClientConfig will not work in query api, use proxyUrl property instead");
         }
 
-        int maxInboundMessageSize = config.getMaxInboundMessageSize() != null
-                ?
-                config.getMaxInboundMessageSize()
-                : Integer.MAX_VALUE;
         nettyChannelBuilder.maxTraceEvents(0)
-                .maxInboundMetadataSize(Integer.MAX_VALUE)
-                .maxInboundMessageSize(maxInboundMessageSize);
+                .maxInboundMetadataSize(Integer.MAX_VALUE);
 
         return FlightGrpcUtils.createFlightClient(new RootAllocator(Long.MAX_VALUE), nettyChannelBuilder.build());
     }
@@ -236,6 +237,16 @@ ProxyDetector createProxyDetector(@Nonnull final String targetUrl, @Nonnull fina
         };
     }
 
+    @Nullable
+    CallOption[] concatCallOptions(@Nullable final CallOption[] base, final CallOption... callOption) {
+        if (base == null || base.length == 0) {
+            return callOption;
+        }
+        List<CallOption> results = new ArrayList<>(List.of(base));
+        Arrays.stream(callOption).filter(Objects::nonNull).forEach(results::add);
+        return results.toArray(new CallOption[0]);
+    }
+
     private static final class FlightSqlIterator implements Iterator<VectorSchemaRoot>, AutoCloseable {
 
         private final List<AutoCloseable> autoCloseable = new ArrayList<>();