Update Strimzi from 0.109.2 to 0.112.0.

Additionaly:
- Update code to compile when using this version (do not use withKraft method which was removed in strimzi/test-container#130)
- Update configurations for tests, use newer versions of containers
- Remove conditional enabling of Kraft mode, since it is on by default

Author: fedinskiy

bom/application/pom.xml

@@ -143,7 +143,7 @@
         <kafka3.version>4.0.0</kafka3.version>
         <lz4.version>1.8.0</lz4.version> <!-- dependency of the kafka-clients that could be overridden by other imported BOMs in the platform -->
         <snappy.version>1.1.10.5</snappy.version>
-        <strimzi-test-container.version>0.109.2</strimzi-test-container.version>
+        <strimzi-test-container.version>0.112.0</strimzi-test-container.version>
         <!-- Scala is used by Kafka so we need to choose a compatible version -->
         <scala.version>2.13.16</scala.version>
         <aws-lambda-java.version>1.4.0</aws-lambda-java.version>

bom/test/pom.xml

@@ -20,7 +20,7 @@
         <jaxb-api.version>2.3.1</jaxb-api.version>
 
         <rxjava1.version>1.3.8</rxjava1.version>
-        <strimzi-test-container.version>0.109.2</strimzi-test-container.version>
+        <strimzi-test-container.version>0.112.0</strimzi-test-container.version>
 
         <opentelemetry-proto.version>1.3.2-alpha</opentelemetry-proto.version>
     </properties>

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/DevServicesKafkaProcessor.java

@@ -109,8 +109,8 @@ private Startable createContainer(DevServicesComposeProjectBuildItem composeProj
                     .withSharedServiceLabel(launchMode.getLaunchMode(), config.serviceName());
             case STRIMZI -> {
                 StrimziKafkaContainer strimzi = new StrimziKafkaContainer(config.effectiveImageName())
+                        .withNodeId(1)
                         .withBrokerId(1)
-                        .withKraft()
                         .waitForRunning();
                 String hostName = ConfigureUtil.configureNetwork(strimzi,
                         composeProjectBuildItem.getDefaultNetworkId(), useSharedNetwork, "kafka");

integration-tests/kafka-devservices/src/test/java/io/quarkus/it/kafka/continuoustesting/DevServicesContainerSharingTest.java

@@ -33,8 +33,7 @@ public class DevServicesContainerSharingTest extends BaseDevServiceTest {
 
     @BeforeAll
     static void beforeAll() {
-        kafka = new StrimziKafkaContainer().withKraft()
-                .withLabel("quarkus-dev-service-kafka", "kafka");
+        kafka = new StrimziKafkaContainer().withLabel("quarkus-dev-service-kafka", "kafka");
         kafka.start();
     }
 

integration-tests/kafka-oauth-keycloak/src/test/java/io/quarkus/it/kafka/KafkaKeycloakTestResource.java

@@ -28,10 +28,10 @@ public Map<String, String> start() {
         client.createRealmFromPath(KEYCLOAK_REALM_JSON);
 
         //Start kafka container
-        this.kafka = new StrimziKafkaContainer("quay.io/strimzi/kafka:latest-kafka-3.7.0")
+        this.kafka = new StrimziKafkaContainer("quay.io/strimzi/kafka:latest-kafka-4.1.0")
                 .withBrokerId(1)
                 .withKafkaConfigurationMap(Map.ofEntries(
-                        entry("listener.security.protocol.map", "JWT:SASL_PLAINTEXT,BROKER1:PLAINTEXT"),
+                        entry("listener.security.protocol.map", "JWT:SASL_PLAINTEXT,BROKER1:PLAINTEXT,CONTROLLER:PLAINTEXT"),
                         entry("listener.name.jwt.oauthbearer.sasl.jaas.config",
                                 getOauthSaslJaasConfig(keycloak.getInternalUrl(), keycloak.getServerUrl())),
                         entry("listener.name.jwt.plain.sasl.jaas.config",

integration-tests/kafka-sasl-elytron/src/test/java/io/quarkus/it/kafka/KafkaSaslTestResource.java

@@ -39,7 +39,8 @@ public Map<String, String> start() {
                 .withBootstrapServers(
                         c -> String.format("SASL_PLAINTEXT://%s:%s", c.getHost(), c.getMappedPort(KAFKA_PORT)))
                 .withKafkaConfigurationMap(Map.ofEntries(
-                        entry("listener.security.protocol.map", "SASL_PLAINTEXT:SASL_PLAINTEXT,BROKER1:PLAINTEXT"),
+                        entry("listener.security.protocol.map",
+                                "SASL_PLAINTEXT:SASL_PLAINTEXT,BROKER1:PLAINTEXT,CONTROLLER:PLAINTEXT"),
                         entry("inter.broker.listener.name", "SASL_PLAINTEXT"),
                         entry("sasl.enabled.mechanisms", "GSSAPI"),
                         entry("sasl.mechanism.inter.broker.protocol", "GSSAPI"),

integration-tests/kafka-sasl/src/test/java/io/quarkus/it/kafka/KafkaSASLTestResource.java

@@ -17,7 +17,7 @@ public class KafkaSASLTestResource implements QuarkusTestResourceLifecycleManage
                     c.getMappedPort(KAFKA_PORT)))
             .withKafkaConfigurationMap(Map.ofEntries(
                     entry("listener.security.protocol.map",
-                            "SASL_PLAINTEXT:SASL_PLAINTEXT,BROKER1:PLAINTEXT,PLAINTEXT:PLAINTEXT"),
+                            "SASL_PLAINTEXT:SASL_PLAINTEXT,BROKER1:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT"),
                     entry("sasl.enabled.mechanisms", "PLAIN"),
                     entry("sasl.mechanism.inter.broker.protocol", "PLAIN"),
                     entry("listener.name.sasl_plaintext.plain.sasl.jaas.config",

integration-tests/kafka-ssl/src/test/java/io/quarkus/it/kafka/KafkaSSLTestResource.java

@@ -28,7 +28,8 @@ public class KafkaSSLTestResource implements QuarkusTestResourceLifecycleManager
                     entry("ssl.truststore.password", "Z_pkTh9xgZovK4t34cGB2o6afT4zZg0L"),
                     entry("ssl.truststore.type", "PKCS12"),
                     entry("ssl.endpoint.identification.algorithm", ""),
-                    entry("listener.security.protocol.map", "BROKER1:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL")))
+                    entry("listener.security.protocol.map",
+                            "BROKER1:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,CONTROLLER:PLAINTEXT")))
             .withCopyFileToContainer(MountableFile.forHostPath("target/certs/kafka-keystore.p12"),
                     "/opt/kafka/config/kafka-keystore.p12")
             .withCopyFileToContainer(MountableFile.forHostPath("target/certs/kafka-truststore.p12"),

integration-tests/kafka-streams/src/test/java/io/quarkus/it/kafka/streams/KafkaSSLTestResource.java

@@ -24,7 +24,8 @@ public class KafkaSSLTestResource implements QuarkusTestResourceLifecycleManager
                     entry("ssl.truststore.password", "Z_pkTh9xgZovK4t34cGB2o6afT4zZg0L"),
                     entry("ssl.truststore.type", "PKCS12"),
                     entry("ssl.endpoint.identification.algorithm=", ""),
-                    entry("listener.security.protocol.map", "BROKER1:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL")))
+                    entry("listener.security.protocol.map",
+                            "BROKER1:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,CONTROLLER:PLAINTEXT")))
             .withBootstrapServers(c -> String.format("SSL://%s:%s", c.getHost(), c.getMappedPort(KAFKA_PORT)))
             .withCopyFileToContainer(MountableFile.forClasspathResource("ks-keystore.p12"),
                     "/opt/kafka/config/kafka-keystore.p12")

test-framework/kafka-companion/src/main/java/io/quarkus/test/kafka/KafkaCompanionResource.java

@@ -15,7 +15,6 @@ public class KafkaCompanionResource implements QuarkusTestResourceLifecycleManag
 
     public static String STRIMZI_KAFKA_IMAGE_KEY = "strimzi.kafka.image";
     public static String KAFKA_PORT_KEY = "kafka.port";
-    public static String KRAFT_KEY = "kraft";
 
     protected String strimziKafkaContainerImage;
     protected Integer kafkaPort;
@@ -56,18 +55,15 @@ public void init(Map<String, String> initArgs) {
             strimziKafkaContainerImage = initArgs.get(STRIMZI_KAFKA_IMAGE_KEY);
             String portString = initArgs.get(KAFKA_PORT_KEY);
             kafkaPort = portString == null ? null : Integer.parseInt(portString);
-            kraft = Boolean.parseBoolean(initArgs.get(KRAFT_KEY));
-            kafka = createContainer(strimziKafkaContainerImage);
-            if (kraft) {
-                kafka.withBrokerId(1).withKraft();
-            }
+            kafka = createContainer(strimziKafkaContainerImage)
+                    .withNodeId(1)
+                    .withBrokerId(1);
             if (kafkaPort != null) {
                 kafka.withPort(kafkaPort);
             }
             Map<String, String> configMap = new HashMap<>(initArgs);
             configMap.remove(STRIMZI_KAFKA_IMAGE_KEY);
             configMap.remove(KAFKA_PORT_KEY);
-            configMap.remove(KRAFT_KEY);
             kafka.withKafkaConfigurationMap(configMap);
         }
     }