KAFKA-16437 Upgrade to Jakarta and Jetty 12 (KIP-1032) (#16754)

This commit implements the changes for KIP-1032. This updates Kafka to Jakarta specs, JavaEE 10 and Jetty 12. The changes here primarily effect Kafka Connect and MM2.

Todo/Notes:

1) I bumped the connect modules to JDK 17 but I also had to bump a couple other things that had a dependency on conect. The tools project depends on connect so that had to be bumped, and streams depends on tools so that needed to be bumped. This means we may need to separate some things if we don't want to enforce JDK 17 on streams.

2) There is an issue with a test in DedicatedMirrorIntegrationTest that I had to change for now that involves escaping characters and not quite sure what to do about it yet. The cause is the Servlet 6 spec changing what is allowed in the path. See: Jetty 12: 400: Ambiguous URI path encoding for path <%=FOO%>~1 (encoded: %3C%25%3DFOO%25%3E%7E1) jetty/jetty.project#11890

3) I had to configure the idle timeout in Jetty requests to match our request timeout so tests didn't fail. This was needed to fix the ConnectWorkerIntegrationTest#testPollTimeoutExpiry() test

Testing is being done by just using the existing tests for Connect and MM2 which should be sufficient.

Reviewers: Greg Harris <[email protected]>, David Arthur <[email protected]>, Chia-Ping Tsai <[email protected]>

LICENSE-binary

@@ -302,7 +302,6 @@ see: licenses/CDDL+GPL-1.1
 javax.activation-api-1.2.0
 javax.annotation-api-1.3.2
 javax.servlet-api-3.1.0
-javax.ws.rs-api-2.1.1
 jaxb-api-2.3.1
 activation-1.1.1
 

build.gradle

@@ -1835,7 +1835,7 @@ project(':generator') {
     implementation libs.argparse4j
     implementation libs.jacksonDatabind
     implementation libs.jacksonJDK8Datatypes
-    implementation libs.jacksonJaxrsJsonProvider
+    implementation libs.jacksonJakartarsJsonProvider
 
     implementation 'org.eclipse.jgit:org.eclipse.jgit:6.4.0.202211300538-r'
     // SSH support for JGit based on Apache MINA sshd
@@ -1882,7 +1882,7 @@ project(':clients') {
     compileOnly libs.jose4j          // for SASL/OAUTHBEARER JWT validation; only used by broker
 
     testImplementation libs.bcpkix
-    testImplementation libs.jacksonJaxrsJsonProvider
+    testImplementation libs.jacksonJakartarsJsonProvider
     testImplementation libs.jose4j
     testImplementation libs.junitJupiter
     testImplementation libs.reload4j
@@ -2446,7 +2446,7 @@ project(':tools') {
     implementation libs.re2j
 
     implementation libs.jose4j                    // for SASL/OAUTHBEARER JWT validation
-    implementation libs.jacksonJaxrsJsonProvider
+    implementation libs.jacksonJakartarsJsonProvider
 
     testImplementation project(':clients')
     testImplementation project(':clients').sourceSets.test.output
@@ -2515,14 +2515,20 @@ project(':trogdor') {
     implementation libs.slf4jApi
     runtimeOnly libs.reload4j
 
-    implementation libs.jacksonJaxrsJsonProvider
+    implementation libs.jacksonJakartarsJsonProvider
     implementation libs.jerseyContainerServlet
     implementation libs.jerseyHk2
     implementation libs.jaxbApi // Jersey dependency that was available in the JDK before Java 9
     implementation libs.activation // Jersey dependency that was available in the JDK before Java 9
-    implementation libs.jettyServer
-    implementation libs.jettyServlet
-    implementation libs.jettyServlets
+    implementation (libs.jettyServer) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyServlet) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyServlets) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
 
     implementation project(':group-coordinator')
     implementation project(':group-coordinator:group-coordinator-api')
@@ -2572,7 +2578,7 @@ project(':shell') {
     implementation project(':raft')
 
     implementation libs.jose4j                    // for SASL/OAUTHBEARER JWT validation
-    implementation libs.jacksonJaxrsJsonProvider
+    implementation libs.jacksonJakartarsJsonProvider
 
     testImplementation project(':clients')
     testImplementation project(':clients').sourceSets.test.output
@@ -2895,11 +2901,7 @@ project(':streams:examples') {
 
   dependencies {
     // this dependency should be removed after we unify data API
-    implementation(project(':connect:json')) {
-      // this transitive dependency is not used in Streams, and it breaks SBT builds
-      exclude module: 'javax.ws.rs-api'
-    }
-
+    implementation(project(':connect:json'))
     implementation project(':streams')
 
     implementation libs.slf4jReload4j
@@ -3342,7 +3344,7 @@ project(':connect:api') {
     api project(':clients')
     implementation libs.slf4jApi
     runtimeOnly libs.reload4j
-    implementation libs.jaxrsApi
+    implementation libs.jakartaRsApi
 
     testImplementation libs.junitJupiter
     testImplementation project(':clients').sourceSets.test.output
@@ -3466,15 +3468,23 @@ project(':connect:runtime') {
     implementation libs.slf4jReload4j
     implementation libs.jose4j                    // for SASL/OAUTHBEARER JWT validation
     implementation libs.jacksonAnnotations
-    implementation libs.jacksonJaxrsJsonProvider
+    implementation libs.jacksonJakartarsJsonProvider
     implementation libs.jerseyContainerServlet
     implementation libs.jerseyHk2
     implementation libs.jaxbApi // Jersey dependency that was available in the JDK before Java 9
     implementation libs.activation // Jersey dependency that was available in the JDK before Java 9
-    implementation libs.jettyServer
-    implementation libs.jettyServlet
-    implementation libs.jettyServlets
-    implementation libs.jettyClient
+    implementation (libs.jettyServer) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyServlet) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyServlets) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyClient) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
     implementation libs.classgraph
     implementation libs.mavenArtifact
     implementation libs.swaggerAnnotations
@@ -3637,7 +3647,7 @@ project(':connect:basic-auth-extension') {
     implementation project(':connect:api')
     implementation libs.slf4jApi
     runtimeOnly libs.reload4j
-    implementation libs.jaxrsApi
+    implementation libs.jakartaRsApi
     implementation libs.jaxAnnotationApi
 
     testImplementation libs.bcpkix
@@ -3683,15 +3693,23 @@ project(':connect:mirror') {
     implementation libs.slf4jApi
     runtimeOnly libs.reload4j
     implementation libs.jacksonAnnotations
-    implementation libs.jacksonJaxrsJsonProvider
+    implementation libs.jacksonJakartarsJsonProvider
     implementation libs.jerseyContainerServlet
     implementation libs.jerseyHk2
     implementation libs.jaxbApi // Jersey dependency that was available in the JDK before Java 9
     implementation libs.activation // Jersey dependency that was available in the JDK before Java 9
-    implementation libs.jettyServer
-    implementation libs.jettyServlet
-    implementation libs.jettyServlets
-    implementation libs.jettyClient
+    implementation (libs.jettyServer) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyServlet) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyServlets) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation (libs.jettyClient) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
     implementation libs.swaggerAnnotations
 
     testImplementation libs.junitJupiter

checkstyle/import-control.xml

@@ -360,8 +360,8 @@
 
   <subpackage name="trogdor">
     <allow pkg="com.fasterxml.jackson" />
-    <allow pkg="javax.servlet" />
-    <allow pkg="javax.ws.rs" />
+    <allow pkg="jakarta.servlet" />
+    <allow pkg="jakarta.ws.rs" />
     <allow pkg="net.sourceforge.argparse4j" />
     <allow pkg="org.apache.kafka.clients" />
     <allow pkg="org.apache.kafka.clients.admin" />
@@ -529,7 +529,7 @@
 
     <subpackage name="rest">
       <allow pkg="org.apache.kafka.connect.health" />
-      <allow pkg="javax.ws.rs" />
+      <allow pkg="jakarta.ws.rs" />
       <allow pkg= "javax.security.auth"/>
       <subpackage name="basic">
         <allow pkg="org.apache.kafka.connect.rest"/>
@@ -555,8 +555,8 @@
       <allow pkg="org.apache.kafka.server.config" />
       <allow pkg="kafka.server" />
       <subpackage name="rest">
-        <allow pkg="javax.ws.rs" />
-        <allow pkg="javax.inject" />
+        <allow pkg="jakarta.ws.rs" />
+        <allow pkg="jakarta.inject" />
         <allow pkg="org.glassfish.jersey" />
         <allow pkg="org.glassfish.hk2" />
       </subpackage>
@@ -571,10 +571,10 @@
 
       <subpackage name="rest">
         <allow pkg="org.eclipse.jetty" />
-        <allow pkg="javax.ws.rs" />
-        <allow pkg="javax.inject" />
+        <allow pkg="jakarta.ws.rs" />
+        <allow pkg="jakarta.inject" />
         <allow pkg="org.glassfish.hk2" />
-        <allow pkg="javax.servlet" />
+        <allow pkg="jakarta.servlet" />
         <allow pkg="org.glassfish.jersey" />
         <allow pkg="com.fasterxml.jackson" />
         <allow pkg="org.apache.http"/>
@@ -588,7 +588,7 @@
       </subpackage>
 
       <subpackage name="distributed">
-        <allow pkg="javax.ws.rs.core" />
+        <allow pkg="jakarta.ws.rs.core" />
       </subpackage>
     </subpackage>
 
@@ -620,8 +620,8 @@
         <allow pkg="org.apache.kafka.common.test" />
         <allow pkg="kafka.zk" />
         <allow pkg="kafka.utils" />
-        <allow class="javax.servlet.http.HttpServletResponse" />
-        <allow class="javax.ws.rs.core.Response" />
+        <allow class="jakarta.servlet.http.HttpServletResponse" />
+        <allow class="jakarta.ws.rs.core.Response" />
         <allow pkg="com.fasterxml.jackson.core.type" />
         <allow pkg="org.apache.kafka.metadata" />
         <allow pkg="org.eclipse.jetty.client"/>
@@ -633,7 +633,7 @@
     <subpackage name="integration">
       <allow pkg="org.apache.kafka.connect.util.clusters" />
       <allow pkg="org.apache.kafka.connect" />
-      <allow pkg="javax.ws.rs" />
+      <allow pkg="jakarta.ws.rs" />
       <allow pkg="org.apache.http"/>
       <allow pkg="org.eclipse.jetty.util"/>
       <!-- for tests -->

connect/api/src/main/java/org/apache/kafka/connect/rest/ConnectRestExtension.java

@@ -51,7 +51,7 @@ public interface ConnectRestExtension extends Configurable, Versioned, Closeable
      * will invoke this method after registering the default Connect resources. If the implementations attempt
      * to re-register any of the Connect resources, it will be ignored and will be logged.
      *
-     * @param restPluginContext The context provides access to JAX-RS {@link javax.ws.rs.core.Configurable} and {@link
+     * @param restPluginContext The context provides access to JAX-RS {@link jakarta.ws.rs.core.Configurable} and {@link
      *                          ConnectClusterState}.The custom JAX-RS resources can be registered via the {@link
      *                          ConnectRestExtensionContext#configurable()}
      */

connect/api/src/main/java/org/apache/kafka/connect/rest/ConnectRestExtensionContext.java

@@ -19,19 +19,20 @@
 
 import org.apache.kafka.connect.health.ConnectClusterState;
 
-import javax.ws.rs.core.Configurable;
+import jakarta.ws.rs.core.Configurable;
+
 
 /**
  * The interface provides the ability for {@link ConnectRestExtension} implementations to access the JAX-RS
- * {@link javax.ws.rs.core.Configurable} and cluster state {@link ConnectClusterState}. The implementation for the interface is provided
+ * {@link jakarta.ws.rs.core.Configurable} and cluster state {@link ConnectClusterState}. The implementation for the interface is provided
  * by the Connect framework.
  */
 public interface ConnectRestExtensionContext {
 
     /**
-     * Provides an implementation of {@link javax.ws.rs.core.Configurable} that can be used to register JAX-RS resources.
+     * Provides an implementation of {@link jakarta.ws.rs.core.Configurable} that can be used to register JAX-RS resources.
      *
-     * @return the JAX-RS {@link javax.ws.rs.core.Configurable}; never {@code null}
+     * @return the JAX-RS {@link jakarta.ws.rs.core.Configurable}; never {@code null}
      */
     Configurable<? extends Configurable<?>> configurable();
 

connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/BasicAuthSecurityRestExtension.java

@@ -51,7 +51,7 @@
  *</Pre>
  *
  * <p>This is a reference implementation of the {@link ConnectRestExtension} interface. It registers an implementation of {@link
- * javax.ws.rs.container.ContainerRequestFilter} that does JAAS based authentication of incoming Basic Auth credentials. {@link
+ * jakarta.ws.rs.container.ContainerRequestFilter} that does JAAS based authentication of incoming Basic Auth credentials. {@link
  * ConnectRestExtension} implementations are loaded via the plugin class loader using {@link java.util.ServiceLoader} mechanism and hence
  * the packaged jar includes {@code META-INF/services/org.apache.kafka.connect.rest.extension.ConnectRestExtension} with the entry
  * {@code org.apache.kafka.connect.extension.auth.jaas.BasicAuthSecurityRestExtension}

connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java

@@ -42,12 +42,13 @@
 import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginContext;
 import javax.security.auth.login.LoginException;
-import javax.ws.rs.HttpMethod;
-import javax.ws.rs.Priorities;
-import javax.ws.rs.container.ContainerRequestContext;
-import javax.ws.rs.container.ContainerRequestFilter;
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.SecurityContext;
+
+import jakarta.ws.rs.HttpMethod;
+import jakarta.ws.rs.Priorities;
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.ContainerRequestFilter;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.SecurityContext;
 
 @Priority(Priorities.AUTHENTICATION)
 public class JaasBasicAuthFilter implements ContainerRequestFilter {

connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/BasicAuthSecurityRestExtensionTest.java

@@ -31,7 +31,8 @@
 import java.util.function.Supplier;
 
 import javax.security.auth.login.Configuration;
-import javax.ws.rs.core.Configurable;
+
+import jakarta.ws.rs.core.Configurable;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;

connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java

@@ -39,11 +39,12 @@
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.ChoiceCallback;
-import javax.ws.rs.HttpMethod;
-import javax.ws.rs.container.ContainerRequestContext;
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.SecurityContext;
-import javax.ws.rs.core.UriInfo;
+
+import jakarta.ws.rs.HttpMethod;
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.SecurityContext;
+import jakarta.ws.rs.core.UriInfo;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;

connect/mirror/src/main/java/org/apache/kafka/connect/mirror/rest/resources/InternalMirrorResource.java

@@ -24,11 +24,11 @@
 
 import java.util.Map;
 
-import javax.inject.Inject;
-import javax.ws.rs.NotFoundException;
-import javax.ws.rs.Path;
-import javax.ws.rs.core.Context;
-import javax.ws.rs.core.UriInfo;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.NotFoundException;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.UriInfo;
 
 @Path("/{source}/{target}/connectors")
 public class InternalMirrorResource extends InternalClusterResource {

connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java

@@ -280,7 +280,13 @@ public void testMultiNodeCluster() throws Exception {
             // Cluster aliases
             final String a = "A";
             // Use a convoluted cluster name to ensure URL encoding/decoding works
-            final String b = "B- ._~:/?#[]@!$&'()*+;=\"<>%{}|\\^`618";
+            // The servlet 6.0 spec no longer allows some characters such as forward slashes, control characters,
+            // etc. even if they are encoded. Jetty 12 will enforce this and throw a 400 ambiguous error
+            // so the string of characters for the variable "b" has been updated to only include characters
+            // that are valid with the new spec.
+            // See https://jakarta.ee/specifications/servlet/6.0/jakarta-servlet-spec-6.0#uri-path-canonicalization
+            // and specifically the section: "10. Rejecting Suspicious Sequences." for details.
+            final String b = "B-_~:?#[]@!$&'()*+=\"<>{}|^`618";
             final String ab = a + "->" + b;
             final String ba = b + "->" + a;
             final String testTopicPrefix = "test-topic-";

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java

@@ -112,8 +112,9 @@
 
 import javax.crypto.KeyGenerator;
 import javax.crypto.SecretKey;
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.UriBuilder;
+
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.UriBuilder;
 
 import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
 import static org.apache.kafka.common.utils.Utils.UncheckedCloseable;

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/ConnectRestConfigurable.java

@@ -24,8 +24,8 @@
 import java.util.Map;
 import java.util.Objects;
 
-import javax.ws.rs.core.Configurable;
-import javax.ws.rs.core.Configuration;
+import jakarta.ws.rs.core.Configurable;
+import jakarta.ws.rs.core.Configuration;
 
 /**
  * The implementation delegates to {@link ResourceConfig} so that we can handle duplicate

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/ConnectRestExtensionContextImpl.java

@@ -20,7 +20,7 @@
 import org.apache.kafka.connect.health.ConnectClusterState;
 import org.apache.kafka.connect.rest.ConnectRestExtensionContext;
 
-import javax.ws.rs.core.Configurable;
+import jakarta.ws.rs.core.Configurable;
 
 public class ConnectRestExtensionContextImpl implements ConnectRestExtensionContext {