KAFKA-18006: Add 3.9.0 to end-to-end test (streams) (#17800)

This commit adds AK 3.9 to the system tests on trunk. Follow-up of #17797 Reviewers: Chia-Ping Tsai <[email protected]>, Bruno Cadonna <[email protected]>
apache · Nov 15, 2024 · 84fe668 · 84fe668
1 parent f02c28b
commit 84fe668
Show file tree

Hide file tree

Showing 14 changed files with 1,359 additions and 11 deletions.
diff --git a/build.gradle b/build.gradle
@@ -2697,6 +2697,7 @@ project(':streams') {
             ':streams:upgrade-system-tests-36:test',
             ':streams:upgrade-system-tests-37:test',
             ':streams:upgrade-system-tests-38:test',
+            ':streams:upgrade-system-tests-39:test',
             ':streams:examples:test'
     ]
   )
@@ -3244,6 +3245,21 @@ project(':streams:upgrade-system-tests-38') {
   }
 }
 
+project(':streams:upgrade-system-tests-39') {
+  base {
+    archivesName = "kafka-streams-upgrade-system-tests-39"
+  }
+
+  dependencies {
+    testImplementation libs.kafkaStreams_39
+    testRuntimeOnly libs.junitJupiter
+  }
+
+  systemTestLibs {
+    dependsOn testJar
+  }
+}
+
 project(':jmh-benchmarks') {
 
   apply plugin: 'io.github.goooler.shadow'

diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
@@ -108,6 +108,7 @@ versions += [
   kafka_36: "3.6.2",
   kafka_37: "3.7.1",
   kafka_38: "3.8.1",
+  kafka_39: "3.9.0",
   // When updating lz4 make sure the compression levels in org.apache.kafka.common.record.CompressionType are still valid
   lz4: "1.8.0",
   mavenArtifact: "3.9.6",
@@ -208,6 +209,7 @@ libs += [
   kafkaStreams_36: "org.apache.kafka:kafka-streams:$versions.kafka_36",
   kafkaStreams_37: "org.apache.kafka:kafka-streams:$versions.kafka_37",
   kafkaStreams_38: "org.apache.kafka:kafka-streams:$versions.kafka_38",
+  kafkaStreams_39: "org.apache.kafka:kafka-streams:$versions.kafka_39",
   lz4: "org.lz4:lz4-java:$versions.lz4",
   metrics: "com.yammer.metrics:metrics-core:$versions.metrics",
   dropwizardMetrics: "io.dropwizard.metrics:metrics-core:$versions.dropwizardMetrics",

diff --git a/settings.gradle b/settings.gradle
@@ -112,6 +112,7 @@ include 'clients',
     'streams:upgrade-system-tests-36',
     'streams:upgrade-system-tests-37',
     'streams:upgrade-system-tests-38',
+    'streams:upgrade-system-tests-39',
     'tools',
     'tools:tools-api',
     'transaction-coordinator',

diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -423,6 +423,12 @@ public class StreamsConfig extends AbstractConfig {
     @SuppressWarnings("WeakerAccess")
     public static final String UPGRADE_FROM_38 = UpgradeFromValues.UPGRADE_FROM_38.toString();
 
+    /**
+     * Config value for parameter {@link #UPGRADE_FROM_CONFIG "upgrade.from"} for upgrading an application from version {@code 3.9.x}.
+     */
+    @SuppressWarnings("WeakerAccess")
+    public static final String UPGRADE_FROM_39 = UpgradeFromValues.UPGRADE_FROM_39.toString();
+
 
     /**
      * Config value for parameter {@link #PROCESSING_GUARANTEE_CONFIG "processing.guarantee"} for at-least-once processing guarantees.
@@ -789,7 +795,7 @@ public class StreamsConfig extends AbstractConfig {
         UPGRADE_FROM_28 + "\", \"" + UPGRADE_FROM_30 + "\", \"" + UPGRADE_FROM_31 + "\", \"" +
         UPGRADE_FROM_32 + "\", \"" + UPGRADE_FROM_33 + "\", \"" + UPGRADE_FROM_34 + "\", \"" +
         UPGRADE_FROM_35 + "\", \"" + UPGRADE_FROM_36 + "\", \"" + UPGRADE_FROM_37 + "\", \"" +
-        UPGRADE_FROM_38 + "(for upgrading from the corresponding old version).";
+        UPGRADE_FROM_38 + "\", \"" + UPGRADE_FROM_39 + "\", \"" + "(for upgrading from the corresponding old version).";
 
     /** {@code topology.optimization} */
     public static final String TOPOLOGY_OPTIMIZATION_CONFIG = "topology.optimization";

diff --git a/streams/src/main/java/org/apache/kafka/streams/internals/UpgradeFromValues.java b/streams/src/main/java/org/apache/kafka/streams/internals/UpgradeFromValues.java
@@ -40,7 +40,8 @@ public enum UpgradeFromValues {
     UPGRADE_FROM_35("3.5"),
     UPGRADE_FROM_36("3.6"),
     UPGRADE_FROM_37("3.7"),
-    UPGRADE_FROM_38("3.8");
+    UPGRADE_FROM_38("3.8"),
+    UPGRADE_FROM_39("3.9");
 
     private final String value;
 

diff --git a/...n/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java b/...n/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java
@@ -130,6 +130,7 @@ public RebalanceProtocol rebalanceProtocol() {
                 case UPGRADE_FROM_36:
                 case UPGRADE_FROM_37:
                 case UPGRADE_FROM_38:
+                case UPGRADE_FROM_39:
                     // we need to add new version when new "upgrade.from" values become available
 
                     // This config is for explicitly sending FK response to a requested partition
@@ -192,6 +193,7 @@ public int configuredMetadataVersion(final int priorVersion) {
                 case UPGRADE_FROM_36:
                 case UPGRADE_FROM_37:
                 case UPGRADE_FROM_38:
+                case UPGRADE_FROM_39:
                     // we need to add new version when new "upgrade.from" values become available
 
                     // This config is for explicitly sending FK response to a requested partition