feat: query timeout improvement (#265)

* feat: (WIP) add writeTimeout and queryTimeout to ClientConfig.  Properties added.  Next, apply in transports.

* feat: (WIP) adding new writeTimeout and queryTimeout config properties.

* chore: checkstyle cleanup

* tests: set timeouts for timeout tests to shorter value for CircleCI

* tests: improve test of cloned GrpcCallOptions object.

* tests: improve timeout exceeded test.

* docs: update javadoc comments and add timeout example.

* docs: update CHANGELOG.md

Author: karel-rehor

CHANGELOG.md

@@ -2,6 +2,18 @@
 
 ### Features
 
+1. [#265](https://github.com/InfluxCommunity/influxdb3-java/pull/265) Add more precise timeout properties to `ClientConfig`.
+   1. Current property `timeout` is deprecated, as it applies only to the Write API and can be confusing to some users.
+   2. Two new properties are added, along with getters and similar setters in the `ClientConfig.Builder`.
+      1. `writeTimeout` - a `java.time.Duration` that applies only to the Write API.
+      2. `queryTimeout` - a `java.time.Duration` used to calculate deadlines when using the Query API.
+   3. These properties can also be defined when creating a client using environment variables. Respectively:
+      1. `INFLUX_WRITE_TIMEOUT` - a positive integer.  The time unit is in seconds.
+      2. `INFLUX_QUERY_TIMEOUT` - a positive integer.  The time unit is in seconds.
+   4. These properties can also be defined when creating a client using system properties. Respectively:
+      1. `influx.writeTimeout` - a positive integer.  The time unit is in seconds.
+      2. `influx.queryTimeout` - a positive integer.  The time unit is in seconds.
+
 ### CI
 
 1. [#266](https://github.com/InfluxCommunity/influxdb3-java/pull/266) Add tests for arm64 CircleCI.

examples/src/main/java/com/influxdb/v3/TimeoutsExample.java

@@ -0,0 +1,86 @@
+/*
+ * 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;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import com.influxdb.v3.client.InfluxDBClient;
+import com.influxdb.v3.client.PointValues;
+import com.influxdb.v3.client.config.ClientConfig;
+
+/**
+ * This example shows how to set universal timeouts for writes and queries.
+ * <p>
+ * The example depends on the "influxdb3-java" module and this module should be built first
+ * by running "mvn install" in the root directory.
+ */
+public final class TimeoutsExample {
+
+    public static String resolveProperty(final String property, final String fallback) {
+        return System.getProperty(property, System.getenv(property)) == null
+            ? fallback : System.getProperty(property, System.getenv(property));
+    }
+
+    private TimeoutsExample() { }
+
+    public static void main(final String[] args) {
+        // timeout to use for writes.  Experiment with lower values to see timeout exceptions.
+        Duration writeTimeout = Duration.ofMillis(5000L);
+        // timeout to use for queries.  Experiment with lower values to see timeout exceptions.
+        Duration queryTimeout = Duration.ofMillis(5000L);
+
+        String host = resolveProperty("INFLUX_HOST", "http://localhost:8181");
+        String token = resolveProperty("INFLUX_TOKEN", "my-token");
+        String database = resolveProperty("INFLUX_DATABASE", "my-database");
+
+        String measurement = "timeout_example";
+
+        ClientConfig config = new ClientConfig.Builder()
+            .host(host)
+            .token(token.toCharArray())
+            .database(database)
+            .writeTimeout(writeTimeout) // set timeout to be used with the Write API
+            .queryTimeout(queryTimeout) // set timeout to be used with the Query API
+            .build();
+
+        try (InfluxDBClient client = InfluxDBClient.getInstance(config)) {
+            client.writeRecord(String.format("%s,id=0001 temp=30.14,ticks=42i", measurement));
+
+            TimeUnit.SECONDS.sleep(1);
+            String sql = String.format("SELECT * FROM %s ORDER BY time DESC", measurement);
+            try (Stream<PointValues> values = client.queryPoints(sql)) {
+                values.forEach(pv -> {
+                    String sv = measurement + ","
+                        + " id: " + pv.getTag("id") + ","
+                        + " fVal: " + pv.getFloatField("temp") + ","
+                        + " iVal: " + pv.getIntegerField("ticks") + ","
+                        + " " + pv.getTimestamp();
+                    System.out.println(sv);
+                });
+            }
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

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

@@ -54,7 +54,11 @@
  *     <li><code>defaultTags</code> - defaultTags added when writing points to InfluxDB</li>
  *     <li><code>gzipThreshold</code> - threshold when gzip compression is used for writing points to InfluxDB</li>
  *     <li><code>writeNoSync</code> - skip waiting for WAL persistence on write</li>
- *     <li><code>responseTimeout</code> - timeout when connecting to InfluxDB</li>
+ *     <li><code>timeout</code> - <i>deprecated in 1.4.0</i> timeout when connecting to InfluxDB,
+ *     please use more informative properties <code>writeTimeout</code> and <code>queryTimeout</code></li>
+ *     <li><code>writeTimeout</code> - timeout when writing data to InfluxDB</li>
+ *     <li><code>queryTimeout</code> - timeout used to calculate a default gRPC deadline when querying InfluxDB.
+ *     Can be <code>null</code>, in which case queries can potentially run forever.</li>
  *     <li><code>allowHttpRedirects</code> - allow redirects for InfluxDB connections</li>
  *     <li><code>disableServerCertificateValidation</code> -
  *          disable server certificate validation for HTTPS connections
@@ -99,7 +103,10 @@ public final class ClientConfig {
     private final Integer gzipThreshold;
     private final Boolean writeNoSync;
     private final Map<String, String> defaultTags;
+    @Deprecated
     private final Duration timeout;
+    private final Duration writeTimeout;
+    private final Duration queryTimeout;
     private final Boolean allowHttpRedirects;
     private final Boolean disableServerCertificateValidation;
     private final String proxyUrl;
@@ -203,15 +210,42 @@ public Map<String, String> getDefaultTags() {
     }
 
     /**
-     * Gets the default timeout to use for the API calls. Default to '10 seconds'.
+     * Gets the default timeout to use for write API calls.
+     * Defaults to '{@value com.influxdb.v3.client.write.WriteOptions#DEFAULT_WRITE_TIMEOUT} seconds'.
+     * <p>
+     * Deprecated in v1.4.0.  Please use more informative <code>getWriteTimeout()</code>.
      *
-     * @return the default timeout to use for the API calls
+     * @return the default timeout to use for write API calls
+     * @see #getWriteTimeout()
      */
     @Nonnull
+    @Deprecated
     public Duration getTimeout() {
         return timeout;
     }
 
+    /**
+     * Gets the default timeout to use for REST Write API calls.  Default is
+     * {@value com.influxdb.v3.client.write.WriteOptions#DEFAULT_WRITE_TIMEOUT} seconds.
+     *
+     * @return the default timeout to use for REST Write API calls.
+     */
+    @Nonnull
+    public Duration getWriteTimeout() {
+        return writeTimeout;
+    }
+
+    /**
+     * Gets the default timeout Duration to use for calculating a gRPC Deadline when making Query API calls.
+     * Can be null, in which case queries can potentially wait or run forever.
+     *
+     * @return the default timeout Duration to use for Query API calls.
+     */
+    @Nullable
+    public Duration getQueryTimeout() {
+        return queryTimeout;
+    }
+
     /**
      * Gets the automatically following HTTP 3xx redirects. Default to 'false'.
      *
@@ -312,6 +346,8 @@ public boolean equals(final Object o) {
                 && Objects.equals(writeNoSync, that.writeNoSync)
                 && Objects.equals(defaultTags, that.defaultTags)
                 && Objects.equals(timeout, that.timeout)
+                && Objects.equals(writeTimeout, that.writeTimeout)
+                && Objects.equals(queryTimeout, that.queryTimeout)
                 && Objects.equals(allowHttpRedirects, that.allowHttpRedirects)
                 && Objects.equals(disableServerCertificateValidation, that.disableServerCertificateValidation)
                 && Objects.equals(proxy, that.proxy)
@@ -325,7 +361,7 @@ public boolean equals(final Object o) {
     public int hashCode() {
         return Objects.hash(host, Arrays.hashCode(token), authScheme, organization,
                 database, writePrecision, gzipThreshold, writeNoSync,
-                timeout, allowHttpRedirects, disableServerCertificateValidation,
+                timeout, writeTimeout, queryTimeout, allowHttpRedirects, disableServerCertificateValidation,
                 proxy, proxyUrl, authenticator, headers,
                 defaultTags, sslRootsFilePath);
     }
@@ -340,6 +376,8 @@ public String toString() {
                 .add("gzipThreshold=" + gzipThreshold)
                 .add("writeNoSync=" + writeNoSync)
                 .add("timeout=" + timeout)
+                .add("writeTimeout=" + writeTimeout)
+                .add("queryTimeout=" + queryTimeout)
                 .add("allowHttpRedirects=" + allowHttpRedirects)
                 .add("disableServerCertificateValidation=" + disableServerCertificateValidation)
                 .add("proxy=" + proxy)
@@ -366,7 +404,10 @@ public static final class Builder {
         private Integer gzipThreshold;
         private Boolean writeNoSync;
         private Map<String, String> defaultTags;
+        @Deprecated
         private Duration timeout;
+        private Duration writeTimeout;
+        private Duration queryTimeout;
         private Boolean allowHttpRedirects;
         private Boolean disableServerCertificateValidation;
         private ProxySelector proxy;
@@ -496,18 +537,55 @@ public Builder defaultTags(@Nullable final Map<String, String> defaultTags) {
         }
 
         /**
-         * Sets the default timeout to use for the API calls. Default to '10 seconds'.
+         * Sets the default timeout to use for Write API calls. Defaults to
+         * '{@value com.influxdb.v3.client.write.WriteOptions#DEFAULT_WRITE_TIMEOUT} seconds'.
+         * <p>
+         * Deprecated in v1.4.0. This setter is superseded by the clearer <code>writeTimeout()</code>.
          *
-         * @param timeout default timeout to use for the API calls. Default to '10 seconds'.
+         * @param timeout default timeout to use for Write API calls. Default to
+         * ''{@value com.influxdb.v3.client.write.WriteOptions#DEFAULT_WRITE_TIMEOUT} seconds'.
          * @return this
+         *
+         * @see #writeTimeout(Duration writeTimeout)
          */
+        @Deprecated
         @Nonnull
         public Builder timeout(@Nullable final Duration timeout) {
 
             this.timeout = timeout;
             return this;
         }
 
+        /**
+         *  Sets the default writeTimeout to use for Write API calls in the REST client.
+         *  Default is {@value com.influxdb.v3.client.write.WriteOptions#DEFAULT_WRITE_TIMEOUT}
+         *
+         * @param writeTimeout default timeout to use for REST API write calls. Default is
+         * {@value com.influxdb.v3.client.write.WriteOptions#DEFAULT_WRITE_TIMEOUT}
+         * @return - this
+         */
+        @Nonnull
+        public Builder writeTimeout(@Nullable final Duration writeTimeout) {
+
+          this.writeTimeout = writeTimeout;
+          return this;
+        }
+
+        /**
+         * Sets standard query timeout used to calculate a gRPC deadline when making Query API calls.
+         * If <code>null</code>, queries can potentially wait or run forever.
+         *
+         * @param queryTimeout default timeout used to calculate deadline for Query API calls.
+         *                     If <code>null</code>, queries can potentially wait or run forever.
+         *                     Default value is <code>null</code>.
+         * @return this
+         */
+        @Nonnull
+        public Builder queryTimeout(@Nullable final Duration queryTimeout) {
+          this.queryTimeout = queryTimeout;
+          return this;
+        }
+
         /**
          * Sets the automatically following HTTP 3xx redirects. Default to 'false'.
          *
@@ -719,6 +797,16 @@ public ClientConfig build(@Nonnull final Map<String, String> env, final Properti
             if (writeNoSync != null) {
                 this.writeNoSync(Boolean.parseBoolean(writeNoSync));
             }
+            final String writeTimeout = get.apply("INFLUX_WRITE_TIMEOUT", "influx.writeTimeout");
+            if (writeTimeout != null) {
+                long to = Long.parseLong(writeTimeout);
+                this.writeTimeout(Duration.ofSeconds(to));
+            }
+            final String queryTimeout = get.apply("INFLUX_QUERY_TIMEOUT", "influx.queryTimeout");
+            if (queryTimeout != null) {
+                long to = Long.parseLong(queryTimeout);
+                this.queryTimeout(Duration.ofSeconds(to));
+            }
 
             return new ClientConfig(this);
         }
@@ -761,7 +849,11 @@ private ClientConfig(@Nonnull final Builder builder) {
         gzipThreshold = builder.gzipThreshold != null ? builder.gzipThreshold : WriteOptions.DEFAULT_GZIP_THRESHOLD;
         writeNoSync = builder.writeNoSync != null ? builder.writeNoSync : WriteOptions.DEFAULT_NO_SYNC;
         defaultTags = builder.defaultTags;
-        timeout = builder.timeout != null ? builder.timeout : Duration.ofSeconds(10);
+        timeout = builder.timeout != null ? builder.timeout : Duration.ofSeconds(WriteOptions.DEFAULT_WRITE_TIMEOUT);
+        writeTimeout = builder.writeTimeout != null
+            ? builder.writeTimeout : builder.timeout != null
+            ? builder.timeout : Duration.ofSeconds(WriteOptions.DEFAULT_WRITE_TIMEOUT);
+        queryTimeout = builder.queryTimeout;
         allowHttpRedirects = builder.allowHttpRedirects != null ? builder.allowHttpRedirects : false;
         disableServerCertificateValidation = builder.disableServerCertificateValidation != null
                 ? builder.disableServerCertificateValidation : false;

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

@@ -288,13 +288,46 @@ public Builder withMaxInboundMessageSize(@Nonnull final Integer maxInboundMessag
             return this;
         }
 
+        /**
+         * Helper method to clone already existing gRPC options.
+         *
+         * @param grpcCallOptions = options to copy
+         * @return this
+         */
+        public Builder fromGrpcCallOptions(@Nonnull final GrpcCallOptions grpcCallOptions) {
+            if (grpcCallOptions.getDeadline() != null) {
+                this.deadline = grpcCallOptions.getDeadline();
+            }
+            if (grpcCallOptions.getExecutor() != null) {
+                this.executor = grpcCallOptions.getExecutor();
+            }
+            if (grpcCallOptions.getCompressorName() != null) {
+                this.compressorName = grpcCallOptions.getCompressorName();
+            }
+            if (grpcCallOptions.getWaitForReady() != null) {
+                this.waitForReady = grpcCallOptions.getWaitForReady();
+            }
+            if (grpcCallOptions.getMaxInboundMessageSize() != null) {
+                this.maxInboundMessageSize = grpcCallOptions.getMaxInboundMessageSize();
+            }
+            if (grpcCallOptions.getMaxOutboundMessageSize() != null) {
+                this.maxOutboundMessageSize = grpcCallOptions.getMaxOutboundMessageSize();
+            }
+            return this;
+        }
+
         /**
          * Sets the maximum allowed message size acceptable sent to the remote peer.
+         * <p>
+         * Note: this property leads to grpc-java issue 12109 and can lead to the connection hanging indefinitely.
+         * See (<a href="https://github.com/grpc/grpc-java/issues/12109">grpc-java 12109</a>)
          *
          * @param maxOutboundMessageSize The maximum message send size
          * @return this
          */
         public Builder withMaxOutboundMessageSize(@Nonnull final Integer maxOutboundMessageSize) {
+            // TODO remove warning about issue 12109 in javadoc above,
+            //  once 12109 is resolved and dependencies are updated.
             this.maxOutboundMessageSize = maxOutboundMessageSize;
             return this;
         }

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

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -37,6 +38,7 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
+import io.grpc.Deadline;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import org.apache.arrow.flight.CallOption;
@@ -396,6 +398,13 @@ private Stream<VectorSchemaRoot> queryData(@Nonnull final String query,
         Arguments.checkNotNull(parameters, "parameters");
         Arguments.checkNotNull(options, "options");
 
+        if (options.grpcCallOptions().getDeadline() == null && config.getQueryTimeout() != null) {
+            options.setGrpcCallOptions(new GrpcCallOptions.Builder()
+                .fromGrpcCallOptions(options.grpcCallOptions())
+                .withDeadline(Deadline.after(config.getQueryTimeout().toMillis(), TimeUnit.MILLISECONDS))
+                .build());
+        }
+
         if (closed) {
             throw new IllegalStateException("InfluxDBClient has been closed.");
         }
@@ -413,6 +422,7 @@ private Stream<VectorSchemaRoot> queryData(@Nonnull final String query,
         });
 
         CallOption[] callOptions = options.grpcCallOptions().getCallOptions();
+
         return flightSqlClient.execute(
                 query,
                 database,

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

@@ -101,7 +101,7 @@ public void checkServerTrusted(
 
         // timeout and redirects
         HttpClient.Builder builder = HttpClient.newBuilder()
-                .connectTimeout(config.getTimeout())
+                .connectTimeout(config.getWriteTimeout())
                 .followRedirects(config.getAllowHttpRedirects()
                         ? HttpClient.Redirect.NORMAL : HttpClient.Redirect.NEVER);