follow up

Author: mjsax

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -62,7 +62,7 @@ import org.apache.kafka.coordinator.share.ShareCoordinator
 import org.apache.kafka.metadata.{ConfigRepository, MetadataCache}
 import org.apache.kafka.server.{ApiVersionManager, ClientMetricsManager, DelegationTokenManager, ProcessRole}
 import org.apache.kafka.server.authorizer._
-import org.apache.kafka.server.common.{GroupVersion, RequestLocal, TransactionVersion}
+import org.apache.kafka.server.common.{GroupVersion, RequestLocal, StreamsVersion, TransactionVersion}
 import org.apache.kafka.server.config.DelegationTokenManagerConfigs
 import org.apache.kafka.server.share.context.ShareFetchContext
 import org.apache.kafka.server.share.{ErroneousAndValidPartitionData, SharePartitionKey}
@@ -2631,11 +2631,15 @@ class KafkaApis(val requestChannel: RequestChannel,
         }
       }
     }
+  }
 
+  private def streamsGroupVersion(): StreamsVersion = {
+    StreamsVersion.fromFeatureLevel(metadataCache.features.finalizedFeatures.getOrDefault(StreamsVersion.FEATURE_NAME, 0.toShort))
   }
 
   private def isStreamsGroupProtocolEnabled: Boolean = {
-      config.groupCoordinatorRebalanceProtocols.contains(Group.GroupType.STREAMS)
+      config.groupCoordinatorRebalanceProtocols.contains(Group.GroupType.STREAMS) &&
+      streamsGroupVersion().streamsGroupSupported
   }
 
   def handleStreamsGroupHeartbeat(request: RequestChannel.Request): CompletableFuture[Unit] = {

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -383,10 +383,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
       warn(s"Share groups and the new '${GroupType.SHARE}' rebalance protocol are enabled. " +
         "This is part of the early access of KIP-932 and MUST NOT be used in production.")
     }
-    if (protocols.contains(GroupType.STREAMS)) {
-      warn(s"Streams groups and the new '${GroupType.STREAMS}' rebalance protocol are enabled. " +
-        "This is part of the early access of KIP-1071 and MUST NOT be used in production.")
-    }
     protocols
   }
 

core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala

@@ -25,7 +25,7 @@ import org.apache.kafka.common.protocol.ApiKeys
 import org.apache.kafka.common.requests.{ApiVersionsRequest, ApiVersionsResponse, RequestUtils}
 import org.apache.kafka.common.test.ClusterInstance
 import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.server.common.{EligibleLeaderReplicasVersion, GroupVersion, MetadataVersion, ShareVersion, StreamsGroupVersion, TransactionVersion}
+import org.apache.kafka.server.common.{EligibleLeaderReplicasVersion, GroupVersion, MetadataVersion, ShareVersion, StreamsVersion, TransactionVersion}
 import org.apache.kafka.test.TestUtils
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Tag
@@ -89,8 +89,8 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) {
       assertEquals(0, apiVersionsResponse.data().supportedFeatures().find(ShareVersion.FEATURE_NAME).minVersion())
       assertEquals(ShareVersion.SV_1.featureLevel(), apiVersionsResponse.data().supportedFeatures().find(ShareVersion.FEATURE_NAME).maxVersion())
 
-      assertEquals(0, apiVersionsResponse.data().supportedFeatures().find(StreamsGroupVersion.FEATURE_NAME).minVersion())
-      assertEquals(StreamsGroupVersion.SGV_1.featureLevel(), apiVersionsResponse.data().supportedFeatures().find(StreamsGroupVersion.FEATURE_NAME).maxVersion())
+      assertEquals(0, apiVersionsResponse.data().supportedFeatures().find(StreamsVersion.FEATURE_NAME).minVersion())
+      assertEquals(StreamsVersion.SV_1.featureLevel(), apiVersionsResponse.data().supportedFeatures().find(StreamsVersion.FEATURE_NAME).maxVersion())
     }
     val expectedApis = if (cluster.controllerListenerName().toScala.contains(listenerName)) {
       ApiVersionsResponse.collectApis(

server-common/src/main/java/org/apache/kafka/server/common/Feature.java

@@ -48,7 +48,7 @@ public enum Feature {
     GROUP_VERSION(GroupVersion.FEATURE_NAME, GroupVersion.values(), GroupVersion.LATEST_PRODUCTION),
     ELIGIBLE_LEADER_REPLICAS_VERSION(EligibleLeaderReplicasVersion.FEATURE_NAME, EligibleLeaderReplicasVersion.values(), EligibleLeaderReplicasVersion.LATEST_PRODUCTION),
     SHARE_VERSION(ShareVersion.FEATURE_NAME, ShareVersion.values(), ShareVersion.LATEST_PRODUCTION),
-    STREAMS_GROUP_VERSION(StreamsGroupVersion.FEATURE_NAME, StreamsGroupVersion.values(), StreamsGroupVersion.LATEST_PRODUCTION),
+    STREAMS_VERSION(StreamsVersion.FEATURE_NAME, StreamsVersion.values(), StreamsVersion.LATEST_PRODUCTION),
 
     /**
      * Features defined only for unit tests and are not used in production.

server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java

@@ -130,8 +130,15 @@ public enum MetadataVersion {
     // *** SHARE GROUPS BECOME PRODUCTION-READY IN THE FUTURE. ITS DEFINITION ALLOWS A SHARE   ***
     // *** GROUPS FEATURE TO BE DEFINED IN 4.1 BUT TURNED OFF BY DEFAULT, ABLE TO BE TURNED ON ***
     // *** DYNAMICALLY TO TRY OUT THE PREVIEW CAPABILITY.                                      ***
-    IBP_4_2_IV0(29, "4.2", "IV0", false);
+    IBP_4_2_IV0(29, "4.2", "IV0", false),
 
+    // Enables "streams" groups by default for new clusters (KIP-1071).
+    //
+    // *** THIS IS A PLACEHOLDER UNSTABLE VERSION WHICH IS USED TO DEFINE THE POINT AT WHICH     ***
+    // *** STREAMS GROUPS BECOME PRODUCTION-READY IN THE FUTURE. ITS DEFINITION ALLOWS A STREAMS ***
+    // *** GROUPS FEATURE TO BE DEFINED IN 4.1 BUT TURNED OFF BY DEFAULT, ABLE TO BE TURNED ON   ***
+    // *** DYNAMICALLY TO TRY OUT THE EARLY ACCESS CAPABILITY.                                   ***
+    IBP_4_2_IV1(30, "4.2", "IV1", false);
 
     // NOTES when adding a new version:
     //   Update the default version in @ClusterTest annotation to point to the latest version
@@ -151,7 +158,7 @@ public enum MetadataVersion {
      * <strong>Think carefully before you update this value. ONCE A METADATA VERSION IS PRODUCTION,
      * IT CANNOT BE CHANGED.</strong>
      */
-    public static final MetadataVersion LATEST_PRODUCTION = IBP_4_0_IV3;
+    public static final MetadataVersion LATEST_PRODUCTION = IBP_4_0_IV3; // when do we update this to `IBP_4_1_IV2`+
     // If you change the value above please also update
     // LATEST_STABLE_METADATA_VERSION version in tests/kafkatest/version.py
 

server-common/src/main/java/org/apache/kafka/server/common/StreamsGroupVersion.java renamed to server-common/src/main/java/org/apache/kafka/server/common/StreamsVersion.java

@@ -18,24 +18,28 @@
 
 import java.util.Map;
 
-public enum StreamsGroupVersion implements FeatureVersion {
+public enum StreamsVersion implements FeatureVersion {
 
-    // Version 0 does disable "streams" groups (KIP-1071).
-    SGV_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+    // Version 0 keeps "streams" groups disabled (KIP-1071).
+    SV_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
 
     // Version 1 enables "streams" groups (KIP-1071).
-    SGV_1(1, MetadataVersion.IBP_4_1_IV2, Map.of());
+    // Using metadata version IBP_4_2_IV1 disables it in AK 4.1 release, and enables it in AK 4.2 release.
+    //  - in AK 4.1, this can be enabled as "early access [unstable]"
+    //  - in AK 4.2, it is planned to go GA (cf `LATEST_PRODUCTION`)
+    SV_1(1, MetadataVersion.IBP_4_2_IV1, Map.of());
 
-    public static final String FEATURE_NAME = "streams.group.version";
+    public static final String FEATURE_NAME = "streams.version";
 
-    // Disabled by default in 4.1 (early access only).
-    public static final StreamsGroupVersion LATEST_PRODUCTION = SGV_0;
+    // Mark "streams" group as unstable in AK 4.1 release
+    // Needs to be updated to SV_1 in AK 4.2, to mark as stable
+    public static final StreamsVersion LATEST_PRODUCTION = SV_0;
 
     private final short featureLevel;
     private final MetadataVersion bootstrapMetadataVersion;
     private final Map<String, Short> dependencies;
 
-    StreamsGroupVersion(
+    StreamsVersion(
         int featureLevel,
         MetadataVersion bootstrapMetadataVersion,
         Map<String, Short> dependencies
@@ -66,13 +70,13 @@ public Map<String, Short> dependencies() {
     }
 
     public boolean streamsGroupSupported() {
-        return featureLevel >= SGV_1.featureLevel;
+        return featureLevel >= SV_1.featureLevel;
     }
 
-    public static StreamsGroupVersion fromFeatureLevel(short version) {
+    public static StreamsVersion fromFeatureLevel(short version) {
         return switch (version) {
-            case 0 -> SGV_0;
-            case 1 -> SGV_1;
+            case 0 -> SV_0;
+            case 1 -> SV_1;
             default -> throw new RuntimeException("Unknown streams group feature level: " + (int) version);
         };
     }

server/src/test/java/org/apache/kafka/server/BrokerFeaturesTest.java

@@ -30,7 +30,7 @@
 import static org.apache.kafka.server.common.Feature.ELIGIBLE_LEADER_REPLICAS_VERSION;
 import static org.apache.kafka.server.common.Feature.GROUP_VERSION;
 import static org.apache.kafka.server.common.Feature.SHARE_VERSION;
-import static org.apache.kafka.server.common.Feature.STREAMS_GROUP_VERSION;
+import static org.apache.kafka.server.common.Feature.STREAMS_VERSION;
 import static org.apache.kafka.server.common.Feature.TRANSACTION_VERSION;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -100,7 +100,7 @@ public void testDefaultFinalizedFeatures() {
                 GROUP_VERSION.featureName(), GROUP_VERSION.latestTesting(),
                 ELIGIBLE_LEADER_REPLICAS_VERSION.featureName(), ELIGIBLE_LEADER_REPLICAS_VERSION.latestTesting(),
                 SHARE_VERSION.featureName(), SHARE_VERSION.latestTesting(),
-                STREAMS_GROUP_VERSION.featureName(), STREAMS_GROUP_VERSION.latestTesting(),
+                STREAMS_VERSION.featureName(), STREAMS_VERSION.latestTesting(),
                 "kraft.version", (short) 0,
                 "test_feature_1", (short) 4,
                 "test_feature_2", (short) 3,

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java

@@ -75,7 +75,7 @@
 @Timeout(600)
 @Tag("integration")
 public class InternalTopicIntegrationTest {
-    public static final EmbeddedKafkaCluster CLUSTER = EmbeddedKafkaCluster.withStreamsRebalanceProtocol(1);
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
 
     @BeforeAll
     public static void startCluster() throws IOException, InterruptedException {

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java

@@ -51,7 +51,7 @@
 @Timeout(600)
 @Tag("integration")
 public class SmokeTestDriverIntegrationTest {
-    public static final EmbeddedKafkaCluster CLUSTER = EmbeddedKafkaCluster.withStreamsRebalanceProtocol(3);
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3);
     public TestInfo testInfo;
 
     @BeforeAll

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StandbyTaskCreationIntegrationTest.java

@@ -57,7 +57,7 @@
 public class StandbyTaskCreationIntegrationTest {
     private static final int NUM_BROKERS = 1;
 
-    public static final EmbeddedKafkaCluster CLUSTER = EmbeddedKafkaCluster.withStreamsRebalanceProtocol(NUM_BROKERS);
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);
 
     private String safeTestName;
 

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java

@@ -143,16 +143,6 @@ public EmbeddedKafkaCluster(final int numBrokers,
         this.time = new MockTime(mockTimeMillisStart, mockTimeNanoStart);
     }
 
-    public static EmbeddedKafkaCluster withStreamsRebalanceProtocol(final int numBrokers) {
-        return withStreamsRebalanceProtocol(numBrokers, new Properties());
-    }
-
-    public static EmbeddedKafkaCluster withStreamsRebalanceProtocol(final int numBrokers, final Properties props) {
-        props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
-        props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
-        return new EmbeddedKafkaCluster(numBrokers, props);
-    }
-
     public void start() {
         try {
             cluster.format();