Merging 7.9.x into 7.9.0

PULL_REQUEST_TEMPLATE.md

@@ -8,3 +8,4 @@ _Describe the testing strategy. Unit and integration tests are expected for any
 - [ ] Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
 - [ ] Ensure relevant issues are linked (description should include text like "Fixes #<issue number>")
 - [ ] Do these changes have compatibility implications for rollback? If so, ensure that the ksql [command version](https://github.com/confluentinc/ksql/blob/master/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/Command.java#L41) is bumped.
+

ksqldb-rest-app/pom.xml

@@ -203,12 +203,12 @@
         </dependency>
 
         <dependency>
-            <!-- one.ducking is a fork of com.kjetland -->
-            <!-- schema-registry is dependent on one.duckling:mbknor-jackson-jsonschema-java8 -->
+            <!-- io.yokota is a fork of com.kjetland -->
+            <!-- schema-registry is dependent on io.yokota:mbknor-jackson-jsonschema-java8 -->
             <!-- hence all the downstream projects had to use that one only -->
-            <groupId>one.duckling</groupId>
+            <groupId>io.yokota</groupId>
             <artifactId>mbknor-jackson-jsonschema-java8</artifactId>
-            <version>1.0.39.1</version>
+            <version>1.0.39.2</version>
             <scope>test</scope>
             <exclusions>
                 <exclusion>

ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlClient.java

@@ -130,7 +130,8 @@ public KsqlTarget target(final URI server, final Map<String, String> additionalH
     final HttpClient client = isUriTls ? httpTlsClient : httpNonTlsClient;
     return new KsqlTarget(client,
         socketAddressFactory.apply(server.getPort(), server.getHost()), localProperties,
-        authHeader, server.getHost(), additionalHeaders, RequestOptions.DEFAULT_TIMEOUT);
+        authHeader, server.getHost(), server.getPath(), additionalHeaders,
+        RequestOptions.DEFAULT_TIMEOUT);
   }
 
   public KsqlTarget targetHttp2(final URI server) {
@@ -139,7 +140,8 @@ public KsqlTarget targetHttp2(final URI server) {
         () -> new IllegalStateException("Must provide http2 options to use targetHttp2"));
     return new KsqlTarget(client,
         socketAddressFactory.apply(server.getPort(), server.getHost()), localProperties,
-        authHeader, server.getHost(), Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        authHeader, server.getHost(), server.getPath(), Collections.emptyMap(),
+        RequestOptions.DEFAULT_TIMEOUT);
   }
 
   @VisibleForTesting

ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java

@@ -386,7 +386,12 @@ private static URI parseUri(final String serverAddress) {
     try {
       final URL url = new URL(serverAddress);
       if (url.getPort() == -1) {
-        return new URL(serverAddress.concat(":") + url.getDefaultPort()).toURI();
+        return new URL(
+                url.getProtocol(),
+                url.getHost(),
+                url.getDefaultPort(),
+                url.getFile()
+        ).toURI();
       }
       return url.toURI();
     } catch (final Exception e) {

ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlTarget.java

@@ -88,6 +88,7 @@ public final class KsqlTarget {
   private final LocalProperties localProperties;
   private final Optional<String> authHeader;
   private final String host;
+  private final String subPath;
   private final Map<String, String> additionalHeaders;
   private final long timeout;
 
@@ -98,13 +99,15 @@ public final class KsqlTarget {
    * @param localProperties Properties sent with ksql requests
    * @param authHeader Optional auth headers
    * @param host The hostname to use for the request, used to set the host header of the request
+   * @param subPath Optional path that can be provided with server name
    */
   KsqlTarget(
       final HttpClient httpClient,
       final SocketAddress socketAddress,
       final LocalProperties localProperties,
       final Optional<String> authHeader,
       final String host,
+      final String subPath,
       final Map<String, String> additionalHeaders,
       final long timeout
   ) {
@@ -113,25 +116,26 @@ public final class KsqlTarget {
     this.localProperties = requireNonNull(localProperties, "localProperties");
     this.authHeader = requireNonNull(authHeader, "authHeader");
     this.host = host;
+    this.subPath = subPath.replaceAll("/\\z", "");
     this.additionalHeaders = requireNonNull(additionalHeaders, "additionalHeaders");
     this.timeout = timeout;
   }
 
   public KsqlTarget authorizationHeader(final String authHeader) {
     return new KsqlTarget(httpClient, socketAddress, localProperties,
-        Optional.of(authHeader), host, additionalHeaders, timeout);
+        Optional.of(authHeader), host, subPath, additionalHeaders, timeout);
   }
 
   public KsqlTarget properties(final Map<String, ?> properties) {
     return new KsqlTarget(httpClient, socketAddress,
         new LocalProperties(properties),
-        authHeader, host, additionalHeaders, timeout);
+        authHeader, host, subPath, additionalHeaders, timeout);
   }
 
   public KsqlTarget timeout(final long timeout) {
     return new KsqlTarget(httpClient, socketAddress,
         localProperties,
-        authHeader, host, additionalHeaders, timeout);
+        authHeader, host, subPath, additionalHeaders, timeout);
   }
 
   public RestResponse<ServerInfo> getServerInfo() {
@@ -503,7 +507,7 @@ private CompletableFuture<ResponseWithBody> execute(
     options.setServer(socketAddress);
     options.setPort(socketAddress.port());
     options.setHost(host);
-    options.setURI(path);
+    options.setURI(subPath + path);
     options.setTimeout(timeout);
 
     httpClient.request(options, ar -> {

ksqldb-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlTargetTest.java

@@ -54,6 +54,7 @@ public class KsqlTargetTest {
 
   private static final String HOST = "host";
   private static final String QUERY = "SELECT * from RATINGS_TABLE;";
+  private static final String SUB_PATH = "";
 
   @Mock
   private HttpClient httpClient;
@@ -151,7 +152,7 @@ private void expectPostQueryRequestChunkHandler() {
   @Test
   public void shouldPostQueryRequest_chunkHandler() {
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
     executor.submit(this::expectPostQueryRequestChunkHandler);
     assertThatEventually(requestStarted::get, is(true));
 
@@ -167,7 +168,7 @@ public void shouldPostQueryRequest_chunkHandler() {
   @Test
   public void shouldPostQueryRequest_chunkHandler_exception() {
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
     executor.submit(this::expectPostQueryRequestChunkHandler);
 
     assertThatEventually(requestStarted::get, is(true));
@@ -183,7 +184,7 @@ public void shouldPostQueryRequest_chunkHandler_exception() {
   public void shouldPostQueryRequest_chunkHandler_nonOkStatusCode() {
     when(httpClientResponse.statusCode()).thenReturn(BAD_REQUEST.code());
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
     executor.submit(this::expectPostQueryRequestChunkHandler);
 
     assertThatEventually(requestStarted::get, is(true));
@@ -198,7 +199,7 @@ public void shouldPostQueryRequest_chunkHandler_nonOkStatusCode() {
   @Test
   public void shouldPostQueryRequest_chunkHandler_closeEarly() {
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
     executor.submit(this::expectPostQueryRequestChunkHandler);
 
     assertThatEventually(requestStarted::get, is(true));
@@ -216,7 +217,7 @@ public void shouldPostQueryRequest_chunkHandler_closeEarly() {
   public void shouldPostQueryRequest_chunkHandler_closeEarlyWithError() {
     doThrow(new RuntimeException("Error!")).when(httpConnection).close();
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
     executor.submit(this::expectPostQueryRequestChunkHandler);
 
     assertThatEventually(requestStarted::get, is(true));
@@ -233,7 +234,7 @@ public void shouldPostQueryRequest_chunkHandler_closeEarlyWithError() {
   @Test
   public void shouldPostQueryRequest_chunkHandler_closeAfterFinish() {
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
     executor.submit(this::expectPostQueryRequestChunkHandler);
 
     assertThatEventually(requestStarted::get, is(true));
@@ -251,7 +252,7 @@ public void shouldPostQueryRequest_chunkHandler_closeAfterFinish() {
   @Test
   public void shouldPostQueryRequest_chunkHandler_partialMessage() {
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, Collections.emptyMap(), RequestOptions.DEFAULT_TIMEOUT);
     executor.submit(this::expectPostQueryRequestChunkHandler);
 
     assertThatEventually(requestStarted::get, is(true));
@@ -272,7 +273,7 @@ public void shouldSendAdditionalHeadersWithKsqlRequest() {
     // Given:
     final Map<String, String> additionalHeaders = ImmutableMap.of("h1", "v1", "h2", "v2");
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        additionalHeaders, RequestOptions.DEFAULT_TIMEOUT);
+        SUB_PATH, additionalHeaders, RequestOptions.DEFAULT_TIMEOUT);
 
     // When:
     executor.submit(() -> {
@@ -294,7 +295,7 @@ public void shouldSendAdditionalHeadersWithKsqlRequest() {
   public void shouldUseTimeout() {
     // Given:
     ksqlTarget = new KsqlTarget(httpClient, socketAddress, localProperties, authHeader, HOST,
-        ImmutableMap.of(), 300L);
+        SUB_PATH, ImmutableMap.of(), 300L);
 
     // When:
     executor.submit(() -> {

pom.xml

@@ -137,8 +137,6 @@
         <skip.docker.test>true</skip.docker.test>
         <!-- Temporarily disabling this because it is causing failures in packaging but not CI builds. -->
         <!-- <compile.warnings-flag>-Werror</compile.warnings-flag> -->
-        <!-- Only used to provide login module implementation for tests -->
-        <jetty.version>9.4.53.v20231009</jetty.version>
         <git-commit-id-plugin.version>4.9.10</git-commit-id-plugin.version>
         <apache.io.version>2.11.0</apache.io.version>
         <io.confluent.ksql.version>7.9.0-0</io.confluent.ksql.version>