KAFKA-18559: Cleanup FinalizedFeatures

JIRA: KAFKA-18559
Cleanup the zk logic and test in `FinalizedFeatures`

Author: frankvicky

core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala

@@ -558,8 +558,7 @@ class KRaftMetadataCache(
     }
     new FinalizedFeatures(image.features().metadataVersion(),
       finalizedFeatures,
-      image.highestOffsetAndEpoch().offset,
-      true)
+      image.highestOffsetAndEpoch().offset)
   }
 }
 

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala

@@ -88,9 +88,7 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
       new FinalizedFeatures(
         MetadataVersion.latestTesting(),
         Collections.singletonMap(TransactionVersion.FEATURE_NAME, TransactionVersion.TV_2.featureLevel()),
-        0,
-        true
-      )
+        0)
     }
 
     when(metadataCache.metadataVersion())

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala

@@ -75,9 +75,7 @@ class TransactionStateManagerTest {
     new FinalizedFeatures(
       MetadataVersion.latestTesting(),
       Collections.singletonMap(TransactionVersion.FEATURE_NAME, TransactionVersion.TV_2.featureLevel()),
-      0,
-      true
-    )
+      0)
   }
 
   val metrics = new Metrics()
@@ -1337,9 +1335,7 @@ class TransactionStateManagerTest {
       new FinalizedFeatures(
         MetadataVersion.latestTesting(),
         Collections.singletonMap(TransactionVersion.FEATURE_NAME, transactionVersion.featureLevel()),
-        0,
-        true
-      )
+        0)
     }
     val transactionManager = new TransactionStateManager(0, scheduler,
       replicaManager, metadataCache, txnConfig, time, metrics)

core/src/test/scala/unit/kafka/network/ProcessorTest.scala

@@ -36,7 +36,7 @@ class ProcessorTest {
     val requestHeader = RequestTestUtils.serializeRequestHeader(
       new RequestHeader(ApiKeys.INIT_PRODUCER_ID, 0, "clientid", 0))
     val apiVersionManager = new SimpleApiVersionManager(ListenerType.CONTROLLER, true,
-      () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0, true))
+      () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0))
     assertThrows(classOf[InvalidRequestException], (() => Processor.parseRequestHeader(apiVersionManager, requestHeader)): Executable,
       "INIT_PRODUCER_ID with listener type CONTROLLER should throw InvalidRequestException exception")
   }
@@ -46,7 +46,7 @@ class ProcessorTest {
     val requestHeader = RequestTestUtils.serializeRequestHeader(
       new RequestHeader(ApiKeys.PRODUCE, 0, "clientid", 0))
     val apiVersionManager = new SimpleApiVersionManager(ListenerType.BROKER, true,
-      () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0, true))
+      () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0))
     assertThrows(classOf[UnsupportedVersionException], (() => Processor.parseRequestHeader(apiVersionManager, requestHeader)): Executable,
       "PRODUCE v0 should throw UnsupportedVersionException exception")
   }

core/src/test/scala/unit/kafka/network/SocketServerTest.scala

@@ -84,7 +84,7 @@ class SocketServerTest {
   TestUtils.clearYammerMetrics()
 
   private val apiVersionManager = new SimpleApiVersionManager(ListenerType.BROKER, true,
-    () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0, true))
+    () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0))
   var server: SocketServer = _
   val sockets = new ArrayBuffer[Socket]
 

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

@@ -177,7 +177,7 @@ class KafkaApisTest extends Logging {
       enabledApis,
       BrokerFeatures.defaultSupportedFeatures(true),
       true,
-      () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0, true))
+      () => new FinalizedFeatures(MetadataVersion.latestTesting(), Collections.emptyMap[String, java.lang.Short], 0))
 
     when(groupCoordinator.isNewGroupCoordinator).thenReturn(config.isNewGroupCoordinatorEnabled)
     setupFeatures(featureVersions)
@@ -218,9 +218,7 @@ class KafkaApisTest extends Logging {
             featureVersions.map { featureVersion =>
               featureVersion.featureName -> featureVersion.featureLevel.asInstanceOf[java.lang.Short]
             }.toMap.asJava,
-            0,
-            true
-          )
+            0)
         }
 
       case _ => throw new IllegalStateException("Test must set an instance of KRaftMetadataCache")

metadata/src/main/java/org/apache/kafka/metadata/publisher/FeaturesPublisher.java

@@ -57,8 +57,8 @@ public void onMetadataUpdate(
         if (delta.featuresDelta() != null) {
             FinalizedFeatures newFinalizedFeatures = new FinalizedFeatures(newImage.features().metadataVersion(),
                     newImage.features().finalizedVersions(),
-                    newImage.provenance().lastContainedOffset(),
-                    true);
+                    newImage.provenance().lastContainedOffset()
+            );
             if (!newFinalizedFeatures.equals(finalizedFeatures)) {
                 log.info("Loaded new metadata {}.", newFinalizedFeatures);
                 finalizedFeatures = newFinalizedFeatures;

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

@@ -27,25 +27,19 @@ public final class FinalizedFeatures {
     private final long finalizedFeaturesEpoch;
 
     public static FinalizedFeatures fromKRaftVersion(MetadataVersion version) {
-        return new FinalizedFeatures(version, Collections.emptyMap(), -1, true);
+        return new FinalizedFeatures(version, Collections.emptyMap(), -1);
     }
 
     public FinalizedFeatures(
         MetadataVersion metadataVersion,
         Map<String, Short> finalizedFeatures,
-        long finalizedFeaturesEpoch,
-        boolean kraftMode
+        long finalizedFeaturesEpoch
     ) {
+        Objects.requireNonNull(metadataVersion);
         this.metadataVersion = metadataVersion;
         this.finalizedFeatures = new HashMap<>(finalizedFeatures);
         this.finalizedFeaturesEpoch = finalizedFeaturesEpoch;
-        // In KRaft mode, we always include the metadata version in the features map.
-        // In ZK mode, we never include it.
-        if (kraftMode) {
-            this.finalizedFeatures.put(MetadataVersion.FEATURE_NAME, metadataVersion.featureLevel());
-        } else {
-            this.finalizedFeatures.remove(MetadataVersion.FEATURE_NAME);
-        }
+        this.finalizedFeatures.put(MetadataVersion.FEATURE_NAME, metadataVersion.featureLevel());
     }
 
     public MetadataVersion metadataVersion() {

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

@@ -24,27 +24,16 @@
 import static org.apache.kafka.server.common.MetadataVersion.FEATURE_NAME;
 import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
 
 class FinalizedFeaturesTest {
     @Test
     public void testKRaftModeFeatures() {
         FinalizedFeatures finalizedFeatures = new FinalizedFeatures(MINIMUM_KRAFT_VERSION,
-                Collections.singletonMap("foo", (short) 2), 123, true);
+                Collections.singletonMap("foo", (short) 2), 123);
         assertEquals(MINIMUM_KRAFT_VERSION.featureLevel(),
                 finalizedFeatures.finalizedFeatures().get(FEATURE_NAME));
         assertEquals((short) 2,
                 finalizedFeatures.finalizedFeatures().get("foo"));
         assertEquals(2, finalizedFeatures.finalizedFeatures().size());
     }
-
-    @Test
-    public void testZkModeFeatures() {
-        FinalizedFeatures finalizedFeatures = new FinalizedFeatures(MINIMUM_KRAFT_VERSION,
-                Collections.singletonMap("foo", (short) 2), 123, false);
-        assertNull(finalizedFeatures.finalizedFeatures().get(FEATURE_NAME));
-        assertEquals((short) 2,
-                finalizedFeatures.finalizedFeatures().get("foo"));
-        assertEquals(1, finalizedFeatures.finalizedFeatures().size());
-    }
 }