Add Kafka 3.4.1 and 3.5.0 + fix tests (#57)

Signed-off-by: see-quick <[email protected]>
strimzi · Jun 16, 2023 · deb9305 · deb9305
1 parent e14479b
commit deb9305
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 7 deletions.
diff --git a/src/main/resources/kafka_versions.json b/src/main/resources/kafka_versions.json
@@ -5,6 +5,7 @@
     "3.1.2": "quay.io/strimzi-test-container/test-container:latest-kafka-3.1.2",
     "3.2.3": "quay.io/strimzi-test-container/test-container:latest-kafka-3.2.3",
     "3.3.2": "quay.io/strimzi-test-container/test-container:latest-kafka-3.3.2",
-    "3.4.0": "quay.io/strimzi-test-container/test-container:latest-kafka-3.4.0"
+    "3.4.1": "quay.io/strimzi-test-container/test-container:latest-kafka-3.4.1",
+    "3.5.0": "quay.io/strimzi-test-container/test-container:latest-kafka-3.5.0"
   }
 }
diff --git a/src/test/java/io/strimzi/test/container/AbstractIT.java b/src/test/java/io/strimzi/test/container/AbstractIT.java
@@ -34,7 +34,7 @@ protected static Stream<Arguments> retrieveKafkaVersionsFile() throws IOExceptio
 
         for (Iterator<Map.Entry<String, JsonNode>> iter = rootNode.get("kafkaVersions").fields(); iter.hasNext(); ) {
             final Map.Entry<String, JsonNode> fields = iter.next();
-            parameters.add(Arguments.of(fields.getValue().asText()));
+            parameters.add(Arguments.of(fields.getValue().asText(), fields.getKey()));
         }
         return parameters.stream();
     }
@@ -46,4 +46,8 @@ protected void assumeDocker() {
     protected void supportsKraftMode(final String imageName) {
         Assumptions.assumeTrue(!imageName.contains("-kafka-2."));
     }
+
+    protected boolean isLessThanKafka350(final String kafkaVersion) {
+        return KafkaVersionService.KafkaVersion.compareVersions(kafkaVersion, "3.5.0") == -1;
+    }
 }
diff --git a/src/test/java/io/strimzi/test/container/StrimziKafkaKraftContainerIT.java b/src/test/java/io/strimzi/test/container/StrimziKafkaKraftContainerIT.java
@@ -43,7 +43,7 @@ public class StrimziKafkaKraftContainerIT extends AbstractIT {
 
     @ParameterizedTest(name = "testStartContainerWithEmptyConfiguration-{0}")
     @MethodSource("retrieveKafkaVersionsFile")
-    void testStartContainerWithEmptyConfiguration(final String imageName) throws ExecutionException, InterruptedException, TimeoutException {
+    void testStartContainerWithEmptyConfiguration(final String imageName, final String kafkaVersion) throws ExecutionException, InterruptedException, TimeoutException {
         assumeDocker();
         supportsKraftMode(imageName);
 
@@ -57,7 +57,12 @@ void testStartContainerWithEmptyConfiguration(final String imageName) throws Exe
             assertThat(systemUnderTest.getClusterId(), notNullValue());
 
             String logsFromKafka = systemUnderTest.getLogs();
-            assertThat(logsFromKafka, containsString("RaftManager nodeId=1"));
+            if (isLessThanKafka350(kafkaVersion)) {
+                assertThat(logsFromKafka, containsString("RaftManager nodeId=1"));
+            } else {
+                assertThat(logsFromKafka, containsString("ControllerServer id=1"));
+                assertThat(logsFromKafka, containsString("SocketServer listenerType=CONTROLLER, nodeId=1"));
+            }
 
             verify();
 
@@ -70,10 +75,9 @@ void testStartContainerWithEmptyConfiguration(final String imageName) throws Exe
 
     @ParameterizedTest(name = "testStartContainerWithSomeConfiguration-{0}")
     @MethodSource("retrieveKafkaVersionsFile")
-    void testStartContainerWithSomeConfiguration(final String imageName) throws ExecutionException, InterruptedException, TimeoutException {
+    void testStartContainerWithSomeConfiguration(final String imageName, final String kafkaVersion) throws ExecutionException, InterruptedException, TimeoutException {
         assumeDocker();
         supportsKraftMode(imageName);
-
         try {
             Map<String, String> kafkaConfiguration = new HashMap<>();
 
@@ -92,7 +96,12 @@ void testStartContainerWithSomeConfiguration(final String imageName) throws Exec
 
             String logsFromKafka = systemUnderTest.getLogs();
 
-            assertThat(logsFromKafka, containsString("RaftManager nodeId=1"));
+            if (isLessThanKafka350(kafkaVersion)) {
+                assertThat(logsFromKafka, containsString("RaftManager nodeId=1"));
+            } else {
+                assertThat(logsFromKafka, containsString("ControllerServer id=1"));
+                assertThat(logsFromKafka, containsString("SocketServer listenerType=CONTROLLER, nodeId=1"));
+            }
             assertThat(logsFromKafka, containsString("log.cleaner.enable = false"));
             assertThat(logsFromKafka, containsString("log.cleaner.backoff.ms = 1000"));
             assertThat(logsFromKafka, containsString("ssl.enabled.protocols = [TLSv1]"));