feat: add idle timeout (#7556)

* Expose a config to set the server's idle timeout. * Increase the default idle timeout from 10 minutes to 24 hours Fixes: #6970 Reviewers: Almog Gavra, Guozhang Wang
confluentinc · May 20, 2021 · db35b98 · db35b98
1 parent 9e1bb91
commit db35b98
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 3 deletions.
diff --git a/ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/Server.java b/ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/Server.java
@@ -107,6 +107,9 @@ public synchronized void start() {
     if (!deploymentIds.isEmpty()) {
       throw new IllegalStateException("Already started");
     }
+    final int idleConnectionTimeoutSeconds =
+            config.getInt(KsqlRestConfig.IDLE_CONNECTION_TIMEOUT_SECONDS);
+
     this.workerExecutor = vertx.createSharedWorkerExecutor("ksql-workers",
         config.getInt(KsqlRestConfig.WORKER_POOL_SIZE));
     final LoggingRateLimiter loggingRateLimiter = new LoggingRateLimiter(config);
@@ -130,7 +133,8 @@ public synchronized void start() {
         final VertxCompletableFuture<String> vcf = new VertxCompletableFuture<>();
         final ServerVerticle serverVerticle = new ServerVerticle(endpoints,
             createHttpServerOptions(config, listener.getHost(), listener.getPort(),
-                listener.getScheme().equalsIgnoreCase("https"), isInternalListener.orElse(false)),
+                listener.getScheme().equalsIgnoreCase("https"), isInternalListener.orElse(false),
+                idleConnectionTimeoutSeconds),
             this, isInternalListener, pullQueryMetrics, loggingRateLimiter);
         vertx.deployVerticle(serverVerticle, vcf);
         final int index = i;
@@ -285,14 +289,14 @@ private void configureTlsCertReload(final KsqlRestConfig config) {
 
   private static HttpServerOptions createHttpServerOptions(final KsqlRestConfig ksqlRestConfig,
       final String host, final int port, final boolean tls,
-      final boolean isInternalListener) {
+      final boolean isInternalListener, final int idleTimeoutSeconds) {
 
     final HttpServerOptions options = new HttpServerOptions()
         .setHost(host)
         .setPort(port)
         .setReuseAddress(true)
         .setReusePort(true)
-        .setIdleTimeout(10 * 60).setIdleTimeoutUnit(TimeUnit.SECONDS)
+        .setIdleTimeout(idleTimeoutSeconds).setIdleTimeoutUnit(TimeUnit.SECONDS)
         .setPerMessageWebSocketCompressionSupported(true)
         .setPerFrameWebSocketCompressionSupported(true);
 

diff --git a/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestConfig.java b/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestConfig.java
@@ -298,6 +298,21 @@ public class KsqlRestConfig extends AbstractConfig {
           + "many instances as there are cores you want to use, as each instance is single "
           + "threaded.";
 
+  public static final String IDLE_CONNECTION_TIMEOUT_SECONDS =
+          KSQL_CONFIG_PREFIX + "idle.connection.timeout.seconds";
+  public static final int DEFAULT_IDLE_CONNECTION_TIMEOUT_SECONDS = 60 * 60 * 24; // one day
+  public static final String IDLE_CONNECTION_TIMEOUT_SECONDS_DOC =
+      "The timeout for idle connections. A connection is idle if there is no data in either "
+          + "direction on that connection for the duration of the timeout. This includes "
+          + "connections where the client only makes occasional requests as well as connections "
+          + "where the server takes a long time to send back any data. An example of the latter "
+          + "case is when there is a long period with no new results to send back in response to "
+          + "a streaming query. You can decrease this timeout to close connections more "
+          + "aggressively and save server resources, or make it longer to be more tolerant of "
+          + "low data volume use cases. Note: even though the server's idle connection timeout "
+          + "is set to a high value, you may have firewalls or proxies that enforce their own "
+          + "idle connection timeouts.";
+
   public static final String WORKER_POOL_SIZE = KSQL_CONFIG_PREFIX + "worker.pool.size";
   public static final String WORKER_POOL_DOC =
       "Max number of worker threads for executing blocking code";
@@ -631,6 +646,13 @@ public class KsqlRestConfig extends AbstractConfig {
             oneOrMore(),
             Importance.MEDIUM,
             VERTICLE_INSTANCES_DOC
+        ).define(
+            IDLE_CONNECTION_TIMEOUT_SECONDS,
+            Type.INT,
+            DEFAULT_IDLE_CONNECTION_TIMEOUT_SECONDS,
+            oneOrMore(),
+            Importance.LOW,
+            IDLE_CONNECTION_TIMEOUT_SECONDS_DOC
         ).define(
             WORKER_POOL_SIZE,
             Type.INT,