KAFKA-16968: Make 3.8-IV0 a stable MetadataVersion and create 3.9-IV0

clients/src/main/resources/common/message/ListOffsetsRequest.json

@@ -34,9 +34,12 @@
   // Version 7 enables listing offsets by max timestamp (KIP-734).
   //
   // Version 8 enables listing offsets by local log start offset (KIP-405).
-  "validVersions": "0-8",
+  //
+  // Version 9 enables listing offsets by last tiered offset (KIP-1005).
+  "validVersions": "0-9",
   "deprecatedVersions": "0",
   "flexibleVersions": "6+",
+  "latestVersionUnstable": true,
   "fields": [
     { "name": "ReplicaId", "type": "int32", "versions": "0+", "entityType": "brokerId",
       "about": "The broker ID of the requester, or -1 if this request is being made by a normal consumer." },

clients/src/main/resources/common/message/ListOffsetsResponse.json

@@ -34,6 +34,8 @@
   //
   // Version 8 enables listing offsets by local log start offset.
   // This is the earliest log start offset in the local log. (KIP-405).
+  //
+  // Version 9 enables listing offsets by last tiered offset (KIP-1005).
   "validVersions": "0-8",
   "flexibleVersions": "6+",
   "fields": [

core/src/test/java/kafka/test/ClusterTestExtensionsTest.java

@@ -185,7 +185,7 @@ public void testNoAutoStart() {
 
     @ClusterTest
     public void testDefaults(ClusterInstance clusterInstance) {
-        Assertions.assertEquals(MetadataVersion.IBP_4_0_IV0, clusterInstance.config().metadataVersion());
+        Assertions.assertEquals(MetadataVersion.latestTesting(), clusterInstance.config().metadataVersion());
     }
 
     @ClusterTests({

core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala

@@ -73,7 +73,7 @@ object ZkMigrationIntegrationTest {
       MetadataVersion.IBP_3_7_IV2,
       MetadataVersion.IBP_3_7_IV4,
       MetadataVersion.IBP_3_8_IV0,
-      MetadataVersion.IBP_4_0_IV0
+      MetadataVersion.IBP_3_9_IV0
     ).map { mv =>
       val serverProperties = new util.HashMap[String, String]()
       serverProperties.put("inter.broker.listener.name", "EXTERNAL")

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

@@ -47,7 +47,7 @@ object ApiVersionsRequestTest {
     List(ClusterConfig.defaultBuilder()
       .setTypes(java.util.Collections.singleton(Type.ZK))
       .setServerProperties(serverProperties)
-      .setMetadataVersion(MetadataVersion.IBP_4_0_IV0)
+      .setMetadataVersion(MetadataVersion.IBP_3_9_IV0)
       .build()).asJava
   }
 

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

@@ -294,7 +294,7 @@ class ReplicationQuotasTest extends QuorumTestHarness {
     features.add(new BrokerRegistrationRequestData.Feature()
       .setName(MetadataVersion.FEATURE_NAME)
       .setMinSupportedVersion(MetadataVersion.IBP_3_0_IV1.featureLevel())
-      .setMaxSupportedVersion(MetadataVersion.IBP_4_0_IV0.featureLevel()))
+      .setMaxSupportedVersion(MetadataVersion.latestTesting().featureLevel()))
     controllerServer.controller.registerBroker(
       ControllerRequestContextUtil.ANONYMOUS_CONTEXT,
       new BrokerRegistrationRequestData()

metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java

@@ -125,7 +125,7 @@ private static MetadataVersion metadataVersionForPartitionChangeRecordVersion(sh
             case (short) 1:
                 return MetadataVersion.IBP_3_7_IV2;
             case (short) 2:
-                return MetadataVersion.IBP_3_8_IV0;
+                return MetadataVersion.IBP_3_9_IV0;
             default:
                 throw new RuntimeException("Unknown PartitionChangeRecord version " + version);
         }

metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java

@@ -196,7 +196,7 @@ public void testConfigurationOperations() throws Throwable {
         ) {
             controlEnv.activeController().registerBroker(ANONYMOUS_CONTEXT,
                 new BrokerRegistrationRequestData().
-                setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
+                setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
                 setBrokerId(0).
                 setLogDirs(Collections.singletonList(Uuid.fromString("iiaQjkRPQcuMULNII0MUeA"))).
                 setClusterId(logEnv.clusterId())).get();
@@ -240,7 +240,7 @@ public void testDelayedConfigurationOperations() throws Throwable {
         ) {
             controlEnv.activeController().registerBroker(ANONYMOUS_CONTEXT,
                 new BrokerRegistrationRequestData().
-                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
+                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
                     setBrokerId(0).
                     setLogDirs(Collections.singletonList(Uuid.fromString("sTbzRAMnTpahIyIPNjiLhw"))).
                     setClusterId(logEnv.clusterId())).get();
@@ -298,7 +298,7 @@ public void testFenceMultipleBrokers() throws Throwable {
                     new BrokerRegistrationRequestData().
                         setBrokerId(brokerId).
                         setClusterId(active.clusterId()).
-                        setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
+                        setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
                         setIncarnationId(Uuid.randomUuid()).
                         setListeners(listeners));
                 brokerEpochs.put(brokerId, reply.get().epoch());
@@ -380,7 +380,7 @@ public void testUncleanShutdownBroker() throws Throwable {
                 }).
                 setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)).
 
-                setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_8_IV0, "test-provided bootstrap ELR enabled")).
+                setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_9_IV0, "test-provided bootstrap ELR enabled")).
                 build()
         ) {
             ListenerCollection listeners = new ListenerCollection();
@@ -394,7 +394,7 @@ public void testUncleanShutdownBroker() throws Throwable {
                     new BrokerRegistrationRequestData().
                         setBrokerId(brokerId).
                         setClusterId(active.clusterId()).
-                        setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_8_IV0)).
+                        setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_9_IV0)).
                         setIncarnationId(Uuid.randomUuid()).
                         setLogDirs(Collections.singletonList(Uuid.randomUuid())).
                         setListeners(listeners));
@@ -463,7 +463,7 @@ public void testUncleanShutdownBroker() throws Throwable {
                 new BrokerRegistrationRequestData().
                     setBrokerId(brokerToUncleanShutdown).
                     setClusterId(active.clusterId()).
-                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_8_IV0)).
+                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_9_IV0)).
                     setIncarnationId(Uuid.randomUuid()).
                     setLogDirs(Collections.singletonList(Uuid.randomUuid())).
                     setListeners(listeners)).get();
@@ -476,7 +476,7 @@ public void testUncleanShutdownBroker() throws Throwable {
                 new BrokerRegistrationRequestData().
                     setBrokerId(lastKnownElr[0]).
                     setClusterId(active.clusterId()).
-                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_8_IV0)).
+                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_9_IV0)).
                     setIncarnationId(Uuid.randomUuid()).
                     setLogDirs(Collections.singletonList(Uuid.randomUuid())).
                     setListeners(listeners)).get();
@@ -717,7 +717,7 @@ public void testUnregisterBroker() throws Throwable {
                     setBrokerId(0).
                     setClusterId(active.clusterId()).
                     setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
-                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
+                    setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
                     setLogDirs(Collections.singletonList(Uuid.fromString("vBpaRsZVSaGsQT53wtYGtg"))).
                     setListeners(listeners));
             assertEquals(5L, reply.get().epoch());

metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java

@@ -371,7 +371,7 @@ public void testPartitionRegistrationToRecord_ElrShouldBeNullIfEmpty() {
             setPartitionEpoch(0);
         List<UnwritableMetadataException> exceptions = new ArrayList<>();
         ImageWriterOptions options = new ImageWriterOptions.Builder().
-            setMetadataVersion(MetadataVersion.IBP_3_8_IV0).
+            setMetadataVersion(MetadataVersion.IBP_3_9_IV0).
             setLossHandler(exceptions::add).
             build();
         assertEquals(new ApiMessageAndVersion(expectRecord, (short) 2), partitionRegistration.toRecord(topicID, 0, options));

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

@@ -202,11 +202,20 @@ public enum MetadataVersion {
     // Add new fetch request version for KIP-951
     IBP_3_7_IV4(19, "3.7", "IV4", false),
 
+    // New version for the Kafka 3.8.0 release.
+    IBP_3_8_IV0(20, "3.8", "IV0", false),
+
+    //
+    // NOTE: MetadataVersions after this point are unstable and may be changed.
+    // If users attempt to use an unstable MetadataVersion, they will get an error.
+    // Please move this comment when updating the LATEST_PRODUCTION constant.
+    //
+
     // Add ELR related supports (KIP-966).
-    IBP_3_8_IV0(20, "3.8", "IV0", true),
+    IBP_3_9_IV0(21, "3.9", "IV0", true),
 
     // Introduce version 1 of the GroupVersion feature (KIP-848).
-    IBP_4_0_IV0(21, "4.0", "IV0", false);
+    IBP_4_0_IV0(22, "4.0", "IV0", false);
 
     // NOTES when adding a new version:
     //   Update the default version in @ClusterTest annotation to point to the latest version
@@ -232,7 +241,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_3_7_IV4;
+    public static final MetadataVersion LATEST_PRODUCTION = IBP_3_8_IV0;
 
     /**
      * An array containing all of the MetadataVersion entries.
@@ -331,7 +340,7 @@ public boolean isDirectoryAssignmentSupported() {
     }
 
     public boolean isElrSupported() {
-        return this.isAtLeast(IBP_3_8_IV0);
+        return this.isAtLeast(IBP_3_9_IV0);
     }
 
     public boolean isKRaftSupported() {
@@ -456,7 +465,9 @@ public short offsetForLeaderEpochRequestVersion() {
     }
 
     public short listOffsetRequestVersion() {
-        if (this.isAtLeast(IBP_3_5_IV0)) {
+        if (this.isAtLeast(IBP_3_9_IV0)) {
+            return 9;
+        } else if (this.isAtLeast(IBP_3_5_IV0)) {
             return 8;
         } else if (this.isAtLeast(IBP_3_0_IV1)) {
             return 7;

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

@@ -23,8 +23,8 @@ public enum TestFeatureVersion implements FeatureVersion {
 
     // TEST_1 released right before MV 3.7-IVO was released, and it has no dependencies
     TEST_1(1, MetadataVersion.IBP_3_7_IV0, Collections.emptyMap()),
-    // TEST_2 released right before MV 3.8-IVO was released, and it depends on this metadata version
-    TEST_2(2, MetadataVersion.IBP_3_8_IV0, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_8_IV0.featureLevel()));
+    // TEST_2 released right before MV 3.9-IVO was released, and it depends on this metadata version
+    TEST_2(2, MetadataVersion.IBP_3_9_IV0, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_9_IV0.featureLevel()));
 
     private final short featureLevel;
     private final MetadataVersion metadataVersionMapping;

server-common/src/test/java/org/apache/kafka/server/common/FeaturesTest.java

@@ -110,7 +110,7 @@ public void testLatestProductionMapsToLatestMetadataVersion(Features features) {
     @EnumSource(MetadataVersion.class)
     public void testDefaultTestVersion(MetadataVersion metadataVersion) {
         short expectedVersion;
-        if (!metadataVersion.isLessThan(MetadataVersion.IBP_3_8_IV0)) {
+        if (!metadataVersion.isLessThan(MetadataVersion.IBP_3_9_IV0)) {
             expectedVersion = 2;
         } else if (!metadataVersion.isLessThan(MetadataVersion.IBP_3_7_IV0)) {
             expectedVersion = 1;

server-common/src/test/java/org/apache/kafka/server/common/MetadataVersionTest.java

@@ -184,8 +184,11 @@ public void testFromVersionString() {
         assertEquals(IBP_3_7_IV3, MetadataVersion.fromVersionString("3.7-IV3"));
         assertEquals(IBP_3_7_IV4, MetadataVersion.fromVersionString("3.7-IV4"));
 
+        assertEquals(IBP_3_8_IV0, MetadataVersion.fromVersionString("3.8"));
         assertEquals(IBP_3_8_IV0, MetadataVersion.fromVersionString("3.8-IV0"));
 
+        assertEquals(IBP_3_9_IV0, MetadataVersion.fromVersionString("3.9-IV0"));
+
         assertEquals(IBP_4_0_IV0, MetadataVersion.fromVersionString("4.0-IV0"));
     }
 
@@ -247,6 +250,7 @@ public void testShortVersion() {
         assertEquals("3.7", IBP_3_7_IV3.shortVersion());
         assertEquals("3.7", IBP_3_7_IV4.shortVersion());
         assertEquals("3.8", IBP_3_8_IV0.shortVersion());
+        assertEquals("3.9", IBP_3_9_IV0.shortVersion());
         assertEquals("4.0", IBP_4_0_IV0.shortVersion());
     }
 
@@ -297,6 +301,7 @@ public void testVersion() {
         assertEquals("3.7-IV3", IBP_3_7_IV3.version());
         assertEquals("3.7-IV4", IBP_3_7_IV4.version());
         assertEquals("3.8-IV0", IBP_3_8_IV0.version());
+        assertEquals("3.9-IV0", IBP_3_9_IV0.version());
         assertEquals("4.0-IV0", IBP_4_0_IV0.version());
     }
 
@@ -365,7 +370,7 @@ public void testDirectoryAssignmentSupported(MetadataVersion metadataVersion) {
     @ParameterizedTest
     @EnumSource(value = MetadataVersion.class)
     public void testIsElrSupported(MetadataVersion metadataVersion) {
-        assertEquals(metadataVersion.isAtLeast(IBP_3_8_IV0), metadataVersion.isElrSupported());
+        assertEquals(metadataVersion.isAtLeast(IBP_3_9_IV0), metadataVersion.isElrSupported());
     }
 
     @ParameterizedTest

tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java

@@ -146,7 +146,7 @@ public void testDowngradeMetadataVersionWithKRaft(ClusterInstance cluster) {
         );
         // Change expected message to reflect possible MetadataVersion range 1-N (N increases when adding a new version)
         assertEquals("Could not disable metadata.version. Invalid update version 0 for feature " +
-                "metadata.version. Local controller 3000 only supports versions 1-21", commandOutput);
+                "metadata.version. Local controller 3000 only supports versions 1-22", commandOutput);
 
         commandOutput = ToolsTestUtils.captureStandardOut(() ->
                 assertEquals(1, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
@@ -321,4 +321,4 @@ public void testHandleDisableDryRun() {
             "Can not disable metadata.version. Can't downgrade below 4%n" +
             "quux can be disabled."), disableOutput);
     }
-}
+}