refactor: set default max message size in constructor

Author: NguyenHoangSon96

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

@@ -27,13 +27,11 @@
 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;
@@ -130,7 +128,7 @@ Stream<VectorSchemaRoot> execute(@Nonnull final String query,
         }
 
         HeaderCallOption headerCallOption = metadataHeader(headers);
-        CallOption[] callOptionArray = concatCallOptions(callOptions, headerCallOption);
+        CallOption[] callOptionArray = GrpcCallOptions.mergeCallOptions(callOptions, headerCallOption);
 
         Ticket ticket = new Ticket(json.getBytes(StandardCharsets.UTF_8));
         FlightStream stream = client.getStream(ticket, callOptionArray);
@@ -237,16 +235,6 @@ 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<>();

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

@@ -22,9 +22,12 @@
 package com.influxdb.v3.client.internal;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.Executor;
+import java.util.stream.Stream;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
@@ -57,6 +60,22 @@ private GrpcCallOptions(@Nonnull final Builder builder) {
         this.callOptions = builder.callOptions.toArray(new CallOption[0]);
     }
 
+    /**
+     * Merges two arrays of {@link CallOption} into a single array. The method combines the elements
+     * from the baseCallOptions array and the additional callOptions array. If either of the input
+     * arrays is null, it will be treated as an empty array.
+     *
+     * @param baseCallOptions the base array of {@link CallOption} instances, may be null
+     * @param callOptions additional {@link CallOption} instances to be added, may also be null
+     * @return a combined array containing all {@link CallOption} instances from both input arrays
+     */
+    public static CallOption[] mergeCallOptions(@Nullable final CallOption[] baseCallOptions, final CallOption... callOptions) {
+        return Stream.concat(
+                Arrays.stream(Optional.ofNullable(baseCallOptions).orElse(new CallOption[0])),
+                Arrays.stream(Optional.ofNullable(callOptions).orElse(new CallOption[0]))
+        ).toArray(CallOption[]::new);
+    }
+
     /**
      * Returns the absolute deadline for a call.
      *
@@ -178,8 +197,17 @@ public static final class Builder {
         private Integer maxOutboundMessageSize;
         private final List<CallOption> callOptions = new ArrayList<>();
 
+        /**
+         * Constructs a new instance of the Builder with default values.
+         * By default, the maximum inbound message size is set to the largest possible value.
+         */
+        public Builder() {
+            this.maxInboundMessageSize = Integer.MAX_VALUE;
+        }
+
         /**
          * Sets the absolute deadline for a rpc call.
+         *
          * @param deadline The deadline
          * @return this
          */
@@ -191,6 +219,7 @@ public Builder withDeadline(final @Nonnull Deadline deadline) {
         /**
          * Sets an {@code executor} to be used instead of the default
          * executor specified with {@link ManagedChannelBuilder#executor}.
+         *
          * @param executor The executor
          * @return this
          */
@@ -206,6 +235,7 @@ public Builder withExecutor(@Nonnull final Executor executor) {
          * <p>It is only safe to call this if the server supports the compression format chosen. There is
          * no negotiation performed; if the server does not support the compression chosen, the call will
          * fail.
+         *
          * @param compressorName The compressor name
          * @return this
          */
@@ -220,6 +250,7 @@ public Builder withCompressorName(@Nonnull final String compressorName) {
          * available. This may dramatically increase the latency of the RPC, but avoids failing
          * "unnecessarily." The default queues the RPC until an attempt to connect has completed, but
          * fails RPCs without sending them if unable to connect.
+         *
          * @return this
          */
         public Builder withWaitForReady() {
@@ -230,6 +261,7 @@ public Builder withWaitForReady() {
         /**
          * Sets the maximum allowed message size acceptable from the remote peer.  If unset, this will
          * default to the value set on the {@link ManagedChannelBuilder#maxInboundMessageSize(int)}.
+         *
          * @param maxInboundMessageSize The max receive message size
          * @return this
          */
@@ -240,6 +272,7 @@ public Builder withMaxInboundMessageSize(@Nonnull final Integer maxInboundMessag
 
         /**
          * Sets the maximum allowed message size acceptable sent to the remote peer.
+         *
          * @param maxOutboundMessageSize The maximum message send size
          * @return this
          */

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

@@ -65,7 +65,6 @@ public final class InfluxDBClientImpl implements InfluxDBClient {
     private static final String DATABASE_REQUIRED_MESSAGE = "Please specify the 'Database' as a method parameter "
             + "or use default configuration at 'ClientConfig.database'.";
 
-    private static final CallOption[] EMPTY_CALL_OPTIONS = new CallOption[0];
     private static final Map<String, Object> NO_PARAMETERS = Map.of();
     private static final List<Class<?>> ALLOWED_NAMED_PARAMETER_TYPES = List.of(
             String.class,
@@ -345,41 +344,22 @@ private Stream<VectorSchemaRoot> queryData(@Nonnull final String query,
             }
         });
 
-        CallOption[] queryCallOptions = createQueryCallOptions(options);
+        CallOption[] callOptions = options.grpcCallOption() != null ? options.grpcCallOption().getCallOptions() : null;
         return flightSqlClient.execute(
                 query,
                 database,
                 options.queryTypeSafe(),
                 parameters,
                 options.headersSafe(),
-                queryCallOptions
+                callOptions
         );
     }
 
-    /**
-     * Creates an array of CallOption with some default CallOption.
-     *
-     * @param options the QueryOptions object
-     * @return the array of CallOption
-     */
-    @Nonnull
-    CallOption[] createQueryCallOptions(@Nonnull final QueryOptions options) {
-        GrpcCallOptions grpcCallOption = options.grpcCallOption();
-        CallOption[] callOptions = grpcCallOption != null ? grpcCallOption.getCallOptions() : EMPTY_CALL_OPTIONS;
-        if (grpcCallOption == null || grpcCallOption.getMaxInboundMessageSize() == null) {
-            callOptions = Stream.concat(
-                    Stream.of(maxInboundMessageCallOption()),
-                    Stream.of(callOptions))
-                    .toArray(CallOption[]::new);
-        }
-        return callOptions;
-    }
-
     @Nonnull
     private CallOption maxInboundMessageCallOption() {
         return new CallOptions.GrpcCallOption() {
             @Override
-            public <T extends AbstractStub<T>> T wrapStub(T stub) {
+            public <T extends AbstractStub<T>> T wrapStub(final T stub) {
                 return stub.withMaxInboundMessageSize(Integer.MAX_VALUE);
             }
         };

src/test/java/com/influxdb/v3/client/internal/FlightSqlClientTest.java

@@ -26,20 +26,16 @@
 import java.util.Map;
 
 import io.grpc.HttpConnectProxiedSocketAddress;
-import io.grpc.Metadata;
 import io.grpc.ProxyDetector;
 import io.grpc.internal.GrpcUtil;
 import org.apache.arrow.flight.CallHeaders;
 import org.apache.arrow.flight.CallInfo;
-import org.apache.arrow.flight.CallOption;
 import org.apache.arrow.flight.CallStatus;
 import org.apache.arrow.flight.FlightClient;
 import org.apache.arrow.flight.FlightClientMiddleware;
 import org.apache.arrow.flight.FlightServer;
-import org.apache.arrow.flight.HeaderCallOption;
 import org.apache.arrow.flight.Location;
 import org.apache.arrow.flight.NoOpFlightProducer;
-import org.apache.arrow.flight.grpc.MetadataAdapter;
 import org.apache.arrow.memory.RootAllocator;
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.AfterEach;
@@ -329,41 +325,6 @@ void createProxyDetector() {
         }
     }
 
-    @Test
-    void concatCallOptions() throws Exception {
-        ClientConfig clientConfig = new ClientConfig.Builder()
-                .host("https://localhost:80")
-                .build();
-        try (FlightSqlClient flightSqlClient = new FlightSqlClient(clientConfig)) {
-            Assertions.assertThatNoException().isThrownBy(() -> {
-                CallOption[] results = flightSqlClient.concatCallOptions(null, null);
-                Assertions.assertThat(results).isNull();
-            });
-
-            MetadataAdapter metadata = new MetadataAdapter(new Metadata());
-            metadata.insert("key1", "value1");
-            HeaderCallOption headerCallOption = new HeaderCallOption(metadata);
-
-            CallOption[] callOptions = flightSqlClient.concatCallOptions(null, headerCallOption);
-            Assertions.assertThat(callOptions).isNotNull();
-            Assertions.assertThat(callOptions.length).isEqualTo(1);
-
-            callOptions = flightSqlClient.concatCallOptions(new CallOption[]{headerCallOption});
-            Assertions.assertThat(callOptions).isNotNull();
-            Assertions.assertThat(callOptions.length).isEqualTo(1);
-
-            GrpcCallOptions grpcCallOption = new GrpcCallOptions.Builder()
-                    .withMaxOutboundMessageSize(1)
-                    .withCompressorName("gzip")
-                    .build();
-
-            callOptions = flightSqlClient.concatCallOptions(grpcCallOption.getCallOptions(), headerCallOption);
-            Assertions.assertThat(callOptions).isNotNull();
-            // This equals to 4 because we always have a default maxInboundMessageSize
-            Assertions.assertThat(callOptions.length).isEqualTo(3);
-        }
-    }
-
     static class CallHeadersMiddleware implements FlightClientMiddleware.Factory {
         CallHeaders headers;
 

src/test/java/com/influxdb/v3/client/internal/GrpcCallOptionsTest.java

@@ -0,0 +1,121 @@
+/*
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package com.influxdb.v3.client.internal;
+
+import io.grpc.stub.AbstractStub;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.CallOptions;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class GrpcCallOptionsTest {
+
+    @Test
+    void testNotSetMaxInboundMessageSize() {
+        GrpcCallOptions grpcCallOptions = new GrpcCallOptions.Builder().build();
+        assertNotNull(grpcCallOptions);
+        assertEquals(Integer.MAX_VALUE, grpcCallOptions.getMaxInboundMessageSize());
+    }
+
+    @Test
+    void testSetMaxInboundMessageSize() {
+        GrpcCallOptions grpcCallOptions = new GrpcCallOptions.Builder()
+                .withMaxInboundMessageSize(2000)
+                .build();
+        assertNotNull(grpcCallOptions);
+        assertEquals(2000, grpcCallOptions.getMaxInboundMessageSize());
+    }
+
+    @Test
+    void testMergeCallOptionsWithBothNonNullArrays() {
+        CallOption option1 = callOption();
+        CallOption option2 = callOption();
+        CallOption[] baseCallOptions = {option1};
+        CallOption[] additionalCallOptions = {option2};
+
+        CallOption[] result = GrpcCallOptions.mergeCallOptions(baseCallOptions, additionalCallOptions);
+
+        assertNotNull(result);
+        assertEquals(2, result.length);
+        assertEquals(option1, result[0]);
+        assertEquals(option2, result[1]);
+    }
+
+    @Test
+    void testMergeCallOptionsWithBaseCallOptionsNull() {
+        CallOption option1 = callOption();
+        CallOption option2 = callOption();
+        CallOption[] baseCallOptions = null;
+        CallOption[] additionalCallOptions = {option1, option2};
+
+        CallOption[] result = GrpcCallOptions.mergeCallOptions(baseCallOptions, additionalCallOptions);
+
+        assertNotNull(result);
+        assertEquals(2, result.length);
+        assertEquals(option1, result[0]);
+        assertEquals(option2, result[1]);
+    }
+
+    @Test
+    void testMergeCallOptionsWithAdditionalCallOptionsNull() {
+        CallOption option1 = callOption();
+        CallOption[] baseCallOptions = {option1};
+        CallOption[] additionalCallOptions = null;
+
+        CallOption[] result = GrpcCallOptions.mergeCallOptions(baseCallOptions, additionalCallOptions);
+
+        assertNotNull(result);
+        assertEquals(1, result.length);
+        assertEquals(option1, result[0]);
+    }
+
+    @Test
+    void testMergeCallOptionsWithBothArraysNull() {
+        CallOption[] result = GrpcCallOptions.mergeCallOptions(null, null);
+
+        assertNotNull(result);
+        assertEquals(0, result.length);
+    }
+
+    @Test
+    void testMergeCallOptionsWithEmptyArrays() {
+        CallOption[] baseCallOptions = {};
+        CallOption[] additionalCallOptions = {};
+
+        CallOption[] result = GrpcCallOptions.mergeCallOptions(baseCallOptions, additionalCallOptions);
+
+        assertNotNull(result);
+        assertEquals(0, result.length);
+    }
+
+    private CallOption callOption() {
+        return new CallOptions.GrpcCallOption() {
+            @Override
+            public <T extends AbstractStub<T>> T wrapStub(final T stub) {
+                return stub.withMaxInboundMessageSize(Integer.MAX_VALUE);
+            }
+        };
+    }
+
+}