HOTIFX: fix Kafka Streams upgrade path from 3.4 to 3.5

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

@@ -284,6 +284,39 @@ public class StreamsConfig extends AbstractConfig {
         OPTIMIZE, NO_OPTIMIZATION, REUSE_KTABLE_SOURCE_TOPICS, MERGE_REPARTITION_TOPICS,
         SINGLE_STORE_SELF_JOIN);
 
+    public enum UpgradeFromValues {
+        UPGRADE_FROM_0100("0.10.0"),
+        UPGRADE_FROM_0101("0.10.1"),
+        UPGRADE_FROM_0102("0.10.2"),
+        UPGRADE_FROM_0110("0.11.0"),
+        UPGRADE_FROM_10("1.0"),
+        UPGRADE_FROM_11("1.1"),
+        UPGRADE_FROM_20("2.0"),
+        UPGRADE_FROM_21("2.1"),
+        UPGRADE_FROM_22("2.2"),
+        UPGRADE_FROM_23("2.3"),
+        UPGRADE_FROM_24("2.4"),
+        UPGRADE_FROM_25("2.5"),
+        UPGRADE_FROM_26("2.6"),
+        UPGRADE_FROM_27("2.7"),
+        UPGRADE_FROM_28("2.8"),
+        UPGRADE_FROM_30("3.0"),
+        UPGRADE_FROM_31("3.1"),
+        UPGRADE_FROM_32("3.2"),
+        UPGRADE_FROM_33("3.3"),
+        UPGRADE_FROM_34("3.4");
+
+        private final String value;
+
+        UpgradeFromValues(final String value) {
+            this.value = value;
+        }
+
+        public String toString() {
+            return value;
+        }
+    }
+
     /**
      * Config value for parameter {@link #UPGRADE_FROM_CONFIG "upgrade.from"} for upgrading an application from version {@code 0.10.0.x}.
      */
@@ -404,6 +437,8 @@ public class StreamsConfig extends AbstractConfig {
     @SuppressWarnings("WeakerAccess")
     public static final String UPGRADE_FROM_34 = "3.4";
 
+    // if we add a new version, we also need to update `UpgradeFromValues` above
+
     /**
      * Config value for parameter {@link #PROCESSING_GUARANTEE_CONFIG "processing.guarantee"} for at-least-once processing guarantees.
      */

streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedSerializer.java

@@ -77,6 +77,7 @@ private static boolean isUpgrade(final Map<String, ?> configs) {
             case StreamsConfig.UPGRADE_FROM_32:
             case StreamsConfig.UPGRADE_FROM_33:
             case StreamsConfig.UPGRADE_FROM_34:
+                // there is no need to add new version here
                 return true;
             default:
                 return false;

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java

@@ -119,6 +119,7 @@ private boolean isNotUpgrade(final Map<String, ?> configs) {
                 case StreamsConfig.UPGRADE_FROM_32:
                 case StreamsConfig.UPGRADE_FROM_33:
                 case StreamsConfig.UPGRADE_FROM_34:
+                    // there is no need to add new versions here
                     return false;
                 default:
                     return true;

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java

@@ -94,6 +94,7 @@ private static boolean upgradeFromV0(final Map<String, ?> configs) {
                 case StreamsConfig.UPGRADE_FROM_31:
                 case StreamsConfig.UPGRADE_FROM_32:
                 case StreamsConfig.UPGRADE_FROM_33:
+                    // there is no need to add new versions hers
                     return true;
                 default:
                     return false;

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java

@@ -584,7 +584,7 @@ public Set<String> makeReady(final Map<String, InternalTopicConfig> topics) {
      */
     // visible for testing
     protected Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final Set<String> topics,
-                                                                final Set<String> tempUnknownTopics) {
+                                                                          final Set<String> tempUnknownTopics) {
         final DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topics);
         final Map<String, KafkaFuture<TopicDescription>> futures = describeTopicsResult.topicNameValues();
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java

@@ -123,6 +123,9 @@ public RebalanceProtocol rebalanceProtocol() {
                 case StreamsConfig.UPGRADE_FROM_31:
                 case StreamsConfig.UPGRADE_FROM_32:
                 case StreamsConfig.UPGRADE_FROM_33:
+                case StreamsConfig.UPGRADE_FROM_34:
+                    // 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
                     // and should not affect the rebalance protocol
                     break;
@@ -178,6 +181,9 @@ public int configuredMetadataVersion(final int priorVersion) {
                 case StreamsConfig.UPGRADE_FROM_31:
                 case StreamsConfig.UPGRADE_FROM_32:
                 case StreamsConfig.UPGRADE_FROM_33:
+                case StreamsConfig.UPGRADE_FROM_34:
+                    // 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
                     // and should not affect the metadata version
                     break;

streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfigurationTest.java

@@ -16,15 +16,34 @@
  */
 package org.apache.kafka.streams.processor.internals.assignment;
 
+import junit.framework.AssertionFailedError;
+import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.streams.StreamsConfig;
+import org.junit.Before;
 import org.junit.Test;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
 import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.EMPTY_RACK_AWARE_ASSIGNMENT_TAGS;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertThrows;
+import static org.mockito.Mockito.mock;
 
 public class AssignorConfigurationTest {
+    private final Map<String, Object> config = new HashMap<>();
+
+    @Before
+    public void setup() {
+        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "app.id");
+        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy");
+        config.put(StreamsConfig.InternalConfig.REFERENCE_CONTAINER_PARTITION_ASSIGNOR, mock(ReferenceContainer.class));
+    }
 
     @Test
     public void configsShouldRejectZeroWarmups() {
@@ -35,4 +54,32 @@ public void configsShouldRejectZeroWarmups() {
 
         assertThat(exception.getMessage(), containsString("Invalid value 0 for configuration max.warmup.replicas"));
     }
+
+    @Test
+    public void rebalanceProtocolShouldSupportAllUpgradeFromVersions() {
+        for (final StreamsConfig.UpgradeFromValues upgradeFrom : StreamsConfig.UpgradeFromValues.values()) {
+            config.put(StreamsConfig.UPGRADE_FROM_CONFIG, upgradeFrom.toString());
+            final AssignorConfiguration assignorConfiguration = new AssignorConfiguration(config);
+
+            try {
+                assignorConfiguration.rebalanceProtocol();
+            } catch (final Exception throwable) {
+                throw new AssertionFailedError("Upgrade from " + upgradeFrom + " failed with " + throwable.getMessage() + "!");
+            }
+        }
+    }
+
+    @Test
+    public void configuredMetadataVersionShouldSupportAllUpgradeFromVersions() {
+        for (final StreamsConfig.UpgradeFromValues upgradeFrom : StreamsConfig.UpgradeFromValues.values()) {
+            config.put(StreamsConfig.UPGRADE_FROM_CONFIG, upgradeFrom.toString());
+            final AssignorConfiguration assignorConfiguration = new AssignorConfiguration(config);
+
+            try {
+                assignorConfiguration.configuredMetadataVersion(0);
+            } catch (final Exception throwable) {
+                throw new AssertionFailedError("Upgrade from " + upgradeFrom + " failed with " + throwable.getMessage() + "!");
+            }
+        }
+    }
 }