feat: allow Java clients to set HTTP/2 multiplex limit (#7871)

confluentinc · Jul 27, 2021 · 790d6fe · 790d6fe
1 parent f631c2b
commit 790d6fe
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 10 deletions.
diff --git a/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/ClientOptions.java b/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/ClientOptions.java
@@ -25,6 +25,7 @@ public interface ClientOptions {
   String DEFAULT_HOST = "localhost";
   int DEFAULT_HOST_PORT = 8088;
   int DEFAULT_EXECUTE_QUERY_MAX_RESULT_ROWS = 10000;
+  int DEFAULT_HTTP2_MULTIPLEXING_LIMIT = -1;
 
   /**
    * Sets the host name of the ksqlDB server to connect to. Defaults to "localhost".
@@ -134,6 +135,14 @@ public interface ClientOptions {
    */
   ClientOptions setExecuteQueryMaxResultRows(int maxRows);
 
+  /**
+   * Sets the maximum number of requests per HTTP/2 connection. Defaults to -1.
+   *
+   * @param http2MultiplexingLimit number of requests
+   * @return a reference to this
+   */
+  ClientOptions setHttp2MultiplexingLimit(int http2MultiplexingLimit);
+
   /**
    * Returns the host name of the ksqlDB server to connect to.
    *
@@ -239,6 +248,13 @@ public interface ClientOptions {
    */
   int getExecuteQueryMaxResultRows();
 
+  /**
+   * Returns the maximum number of requests per HTTP/2 connection.
+   *
+   * @return number of requests
+   */
+  int getHttp2MultiplexingLimit();
+
   /**
    * Creates a copy of these {@code ClientOptions}.
    *
@@ -249,4 +265,4 @@ public interface ClientOptions {
   static ClientOptions create() {
     return new ClientOptionsImpl();
   }
-}
+}
diff --git a/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientImpl.java b/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientImpl.java
@@ -621,7 +621,8 @@ private static HttpClient createHttpClient(final Vertx vertx, final ClientOption
         .setHttp2ClearTextUpgrade(false)
         .setVerifyHost(clientOptions.isVerifyHost())
         .setDefaultHost(clientOptions.getHost())
-        .setDefaultPort(clientOptions.getPort());
+        .setDefaultPort(clientOptions.getPort())
+        .setHttp2MultiplexingLimit(clientOptions.getHttp2MultiplexingLimit());
     if (clientOptions.isUseTls() && !clientOptions.getTrustStore().isEmpty()) {
       final JksOptions jksOptions = VertxSslOptionsFactory.getJksTrustStoreOptions(
           clientOptions.getTrustStore(),
@@ -681,4 +682,4 @@ public String toString() {
         + ", vertx=" + vertx
         + '}';
   }
-}
+}
diff --git a/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientOptionsImpl.java b/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientOptionsImpl.java
@@ -35,6 +35,7 @@ public class ClientOptionsImpl implements ClientOptions {
   private String basicAuthUsername;
   private String basicAuthPassword;
   private int executeQueryMaxResultRows = ClientOptions.DEFAULT_EXECUTE_QUERY_MAX_RESULT_ROWS;
+  private int http2MultiplexingLimit = ClientOptions.DEFAULT_HTTP2_MULTIPLEXING_LIMIT;
 
   /**
    * {@code ClientOptions} should be instantiated via {@link ClientOptions#create}, NOT via this
@@ -52,7 +53,7 @@ private ClientOptionsImpl(
       final String trustStorePath, final String trustStorePassword,
       final String keyStorePath, final String keyStorePassword, final String keyPassword,
       final String keyAlias, final String basicAuthUsername, final String basicAuthPassword,
-      final int executeQueryMaxResultRows) {
+      final int executeQueryMaxResultRows, final int http2MultiplexingLimit) {
     this.host = Objects.requireNonNull(host);
     this.port = port;
     this.useTls = useTls;
@@ -68,6 +69,7 @@ private ClientOptionsImpl(
     this.basicAuthUsername = basicAuthUsername;
     this.basicAuthPassword = basicAuthPassword;
     this.executeQueryMaxResultRows = executeQueryMaxResultRows;
+    this.http2MultiplexingLimit = http2MultiplexingLimit;
   }
 
   @Override
@@ -150,6 +152,12 @@ public ClientOptions setExecuteQueryMaxResultRows(final int maxRows) {
     return this;
   }
 
+  @Override
+  public ClientOptions setHttp2MultiplexingLimit(final int http2MultiplexingLimit) {
+    this.http2MultiplexingLimit = http2MultiplexingLimit;
+    return this;
+  }
+
   @Override
   public String getHost() {
     return host == null ? "" : host;
@@ -225,6 +233,11 @@ public int getExecuteQueryMaxResultRows() {
     return executeQueryMaxResultRows;
   }
 
+  @Override
+  public int getHttp2MultiplexingLimit() {
+    return http2MultiplexingLimit;
+  }
+
   @Override
   public ClientOptions copy() {
     return new ClientOptionsImpl(
@@ -234,7 +247,7 @@ public ClientOptions copy() {
         trustStorePath, trustStorePassword,
         keyStorePath, keyStorePassword, keyPassword, keyAlias,
         basicAuthUsername, basicAuthPassword,
-        executeQueryMaxResultRows);
+        executeQueryMaxResultRows, http2MultiplexingLimit);
   }
 
   // CHECKSTYLE_RULES.OFF: CyclomaticComplexity
@@ -261,14 +274,15 @@ public boolean equals(final Object o) {
         && Objects.equals(keyPassword, that.keyPassword)
         && Objects.equals(keyAlias, that.keyAlias)
         && Objects.equals(basicAuthUsername, that.basicAuthUsername)
-        && Objects.equals(basicAuthPassword, that.basicAuthPassword);
+        && Objects.equals(basicAuthPassword, that.basicAuthPassword)
+        && http2MultiplexingLimit == that.http2MultiplexingLimit;
   }
 
   @Override
   public int hashCode() {
     return Objects.hash(host, port, useTls, verifyHost, useAlpn, trustStorePath,
         trustStorePassword, keyStorePath, keyStorePassword, keyPassword, keyAlias,
-        basicAuthUsername, basicAuthPassword, executeQueryMaxResultRows);
+        basicAuthUsername, basicAuthPassword, executeQueryMaxResultRows, http2MultiplexingLimit);
   }
 
   @Override
@@ -287,7 +301,8 @@ public String toString() {
         + ", keyAlias='" + keyAlias + '\''
         + ", basicAuthUsername='" + basicAuthUsername + '\''
         + ", basicAuthPassword='" + basicAuthPassword + '\''
-        + ", executeQueryMaxResultRows=" + executeQueryMaxResultRows
+        + ", executeQueryMaxResultRows=" + executeQueryMaxResultRows + '\''
+        + ", http2MultiplexingLimit=" + http2MultiplexingLimit
         + '}';
   }
-}
+}
diff --git a/ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/impl/ClientOptionsImplTest.java b/ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/impl/ClientOptionsImplTest.java
@@ -63,7 +63,10 @@ public void shouldImplementHashCodeAndEquals() {
         .addEqualityGroup(
             ClientOptions.create().setExecuteQueryMaxResultRows(10)
         )
+        .addEqualityGroup(
+            ClientOptions.create().setHttp2MultiplexingLimit(5)
+        )
         .testEquals();
   }
 
-}
+}