chore: refactor gRPC Deadline checks and add logger warning for impractical values.

Author: karel-rehor

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

@@ -414,17 +414,24 @@ private Stream<VectorSchemaRoot> queryData(@Nonnull final String query,
             }
         });
 
-        // Copy call options to create a new deadline for this query
-        // if queryTimeout is configured and deadline isn't explicitly set
-        GrpcCallOptions grpcCallOptionsCopy = options.grpcCallOptions();
-        if (grpcCallOptionsCopy.getDeadline() == null && config.getQueryTimeout() != null) {
-            grpcCallOptionsCopy = new GrpcCallOptions.Builder()
-                .fromGrpcCallOptions(grpcCallOptionsCopy)
-                .withDeadline(Deadline.after(config.getQueryTimeout().toMillis(), TimeUnit.MILLISECONDS))
-                .build();
+        GrpcCallOptions.Builder builder = new GrpcCallOptions.Builder()
+            .fromGrpcCallOptions(options.grpcCallOptions());
+
+        if (options.grpcCallOptions().getDeadline() == null) {
+            if (config.getQueryTimeout() != null) {
+                builder.withDeadline(Deadline.after(config.getQueryTimeout().toMillis(), TimeUnit.MILLISECONDS));
+            }
+        } else if (options.grpcCallOptions().getDeadline().timeRemaining(TimeUnit.NANOSECONDS) <= 0) {
+            LOG.warning("Received impractical gRPC call options deadline "
+                + options.grpcCallOptions().getDeadline());
+            if (config.getQueryTimeout() != null) {
+                LOG.warning("Using configuration query timeout "
+                    + config.getQueryTimeout() + " as a replacement.");
+                builder.withDeadline(Deadline.after(config.getQueryTimeout().toMillis(), TimeUnit.MILLISECONDS));
+            }
         }
 
-        CallOption[] callOptions = grpcCallOptionsCopy.getCallOptions();
+        CallOption[] callOptions = builder.build().getCallOptions();
 
         return flightSqlClient.execute(
                 query,

src/test/java/com/influxdb/v3/client/query/QueryOptionsTest.java

@@ -329,6 +329,50 @@ public void queryOptionsUnchangedByCall() throws IOException {
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
+    }
+
+    @Test
+    public void impracticalGRPCDeadlineReplacedByQueryTimeout() throws IOException {
+        int freePort = findFreePort();
+        URI uri = URI.create("http://127.0.0.1:" + freePort);
+        int rowCount = 10;
+        try (VectorSchemaRoot vectorSchemaRoot = TestUtils.generateVectorSchemaRoot(10, rowCount);
+             BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);
+             FlightServer flightServer = TestUtils.simpleFlightServer(uri, allocator,
+                 TestUtils.simpleProducer(vectorSchemaRoot))
+        ) {
+            flightServer.start();
+
+            String host = String.format("http://%s:%d", uri.getHost(), uri.getPort());
+            ClientConfig clientConfig = new ClientConfig.Builder()
+                .host(host)
+                .database("test")
+                .writeTimeout(Duration.ofSeconds(60))
+                .queryTimeout(Duration.ofSeconds(60))
+                .build();
+
+            GrpcCallOptions grpcCallOption = new GrpcCallOptions.Builder()
+                .withMaxInboundMessageSize(1024 * 1024 * 1024)
+                .withDeadline(Deadline.after(-2, TimeUnit.MILLISECONDS))
+                .build();
 
+            QueryOptions queryOptions = new QueryOptions("test");
+            queryOptions.setGrpcCallOptions(grpcCallOption);
+            QueryOptions originalQueryOptions = queryOptions.clone();
+            Assertions.assertThat(originalQueryOptions).isEqualTo(queryOptions);
+
+            try (InfluxDBClient influxDBClient = InfluxDBClient.getInstance(clientConfig)) {
+                Assertions.assertThatNoException().isThrownBy(() -> {
+                    Stream<PointValues> stream = influxDBClient.queryPoints(
+                        "Select * from \"sensors\"",
+                        queryOptions);
+                    Assertions.assertThat(stream.count()).isEqualTo(rowCount);
+                    stream.close();
+                });
+            }
+            Assertions.assertThat(queryOptions).isEqualTo(originalQueryOptions);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
     }
 }