Deprecate and stop using the .status.kafkaMetadataState field (#11890)

Signed-off-by: Jakub Scholz <www@scholzj.com>

Author: scholzj

CHANGELOG.md

@@ -5,6 +5,10 @@
 * Set `blockOwnerDeletion` to `true` in the owner references in Strimzi managed resources.
   Deleting the Strimzi custom resources will now by default wait for the deletion of all the owned Kubernetes resources.
 
+### Major changes, deprecations, and removals
+
+* The `.status.kafkaMetadataState` field in the `Kafka` custom resource is deprecated and not used anymore.
+
 ## 0.48.0
 
 * Add support for Kafka 4.1.0.

api/src/main/java/io/strimzi/api/kafka/model/kafka/KafkaMetadataState.java

@@ -11,6 +11,7 @@
  * Represents where metadata are stored for the current cluster (ZooKeeper or KRaft)
  * or if a migration from ZooKeeper to KRaft is in progress and in which phase
  */
+@Deprecated
 public enum KafkaMetadataState {
     /**
      * The metadata are stored in ZooKeeper.

api/src/main/java/io/strimzi/api/kafka/model/kafka/KafkaStatus.java

@@ -109,11 +109,15 @@ public void setKafkaMetadataVersion(String kafkaMetadataVersion) {
         this.kafkaMetadataVersion = kafkaMetadataVersion;
     }
 
-    @Description("Defines where cluster metadata are stored. Since Kafka KRaft has been fully adopted, the only applicable value is `KRaft`.")
+    @Description("Defines where cluster metadata are stored. This property is deprecated and not used anymore.")
+    @Deprecated
+    @DeprecatedProperty(description = "The `kafkaMetadataState` property is deprecated and will be removed in the future.")
+    @PresentInVersions("v1beta2")
     public KafkaMetadataState getKafkaMetadataState() {
         return kafkaMetadataState;
     }
 
+    @SuppressWarnings("deprecation") // KafkaMetadataState is deprecated
     public void setKafkaMetadataState(KafkaMetadataState metadataState) {
         this.kafkaMetadataState = metadataState;
     }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperator.java

@@ -171,12 +171,6 @@ public Future<KafkaStatus> createOrUpdate(Reconciliation reconciliation, Kafka k
                     status.setKafkaMetadataVersion(kafkaAssembly.getStatus().getKafkaMetadataVersion());
                 }
 
-                if (status.getKafkaMetadataState() == null
-                        && kafkaAssembly.getStatus().getKafkaMetadataState() != null)  {
-                    // Copy the metadata state if needed
-                    status.setKafkaMetadataState(kafkaAssembly.getStatus().getKafkaMetadataState());
-                }
-
                 if (status.getAutoRebalance() == null
                         && kafkaAssembly.getStatus().getAutoRebalance() != null
                         && reconcileState.isAutoRebalancingEnabled()) {
@@ -217,6 +211,9 @@ public Future<KafkaStatus> createOrUpdate(Reconciliation reconciliation, Kafka k
     Future<Void> reconcile(ReconciliationState reconcileState)  {
         Promise<Void> chainPromise = Promise.promise();
 
+        // The .status.kafkaMetadataState field is deprecated and we do not set it anymore. But we can still use it to
+        // check the state in the older resources previously reconciled by other operators. And we can prevent
+        // reconciliation of clusters not yet migrated to KRaft that way.
         if (ReconcilerUtils.nonMigratedCluster(reconcileState.kafkaAssembly)) {
             throw new InvalidConfigurationException("Strimzi " + OPERATOR_VERSION + " supports only KRaft-based Apache Kafka clusters. Please make sure your cluster is migrated to KRaft before using Strimzi " + OPERATOR_VERSION + ".");
         }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaReconciler.java

@@ -14,7 +14,6 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.strimzi.api.kafka.model.common.Condition;
 import io.strimzi.api.kafka.model.kafka.Kafka;
-import io.strimzi.api.kafka.model.kafka.KafkaMetadataState;
 import io.strimzi.api.kafka.model.kafka.KafkaResources;
 import io.strimzi.api.kafka.model.kafka.KafkaStatus;
 import io.strimzi.api.kafka.model.kafka.UsedNodePoolStatus;
@@ -1181,7 +1180,6 @@ protected Future<Void> nodePortExternalListenerStatus() {
     /* test */ Future<Void> updateKafkaStatus(KafkaStatus kafkaStatus) {
         kafkaStatus.setListeners(listenerReconciliationResults.listenerStatuses);
         kafkaStatus.setKafkaVersion(kafka.getKafkaVersion().version());
-        kafkaStatus.setKafkaMetadataState(KafkaMetadataState.KRaft);
 
         return Future.succeededFuture();
     }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/ReconcilerUtils.java

@@ -390,6 +390,7 @@ public static boolean trackedServerCertChanged(Pod pod, Map<Integer, String> cer
      *
      * @return      True ZooKeeper metadata are in use or when the cluster is in migration. False otherwise.
      */
+    @SuppressWarnings("deprecation") // KafkaMetadataState is deprecated, but we still use it to check for clusters not migrated to KRaft
     public static boolean nonMigratedCluster(Kafka kafka) {
         // When the Kafka status or the metadata state are null, we cannot decide anything about KRaft (it can be a new
         // cluster or a cluster that is still doing the first deployment). Only when it is set to one of the non-KRaft

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperatorKRaftMockTest.java

@@ -261,7 +261,6 @@ private void basicCheck()    {
         assertThat(k.getStatus().getKafkaMetadataVersion(), startsWith(VERSIONS.defaultVersion().metadataVersion() + "-IV"));
         assertThat(k.getStatus().getKafkaVersion(), is(VERSIONS.defaultVersion().version()));
         assertThat(k.getStatus().getOperatorLastSuccessfulVersion(), is(KafkaAssemblyOperator.OPERATOR_VERSION));
-        assertThat(k.getStatus().getKafkaMetadataState().toValue(), is("KRaft"));
         assertThat(k.getStatus().getKafkaNodePools().stream().map(UsedNodePoolStatus::getName).toList(), is(List.of("brokers", "controllers")));
     }
 

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperatorWithKRaftTest.java

@@ -1435,6 +1435,7 @@ vertx, new PlatformFeaturesAvailability(false, KUBERNETES_VERSION),
      * @param context   Test context
      */
     @Test
+    @SuppressWarnings("deprecation") // KafkaMetadataState is deprecated
     public void testClusterWithZooKeeperMetadata(VertxTestContext context)  {
         Kafka kafka = new KafkaBuilder(KAFKA)
                 .editStatus()
@@ -1471,6 +1472,7 @@ vertx, new PlatformFeaturesAvailability(false, KUBERNETES_VERSION),
      * @param context   Test context
      */
     @Test
+    @SuppressWarnings("deprecation") // KafkaMetadataState is deprecated
     public void testClusterWithMigrationMetadata(VertxTestContext context)  {
         Kafka kafka = new KafkaBuilder(KAFKA)
                 .editStatus()

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaReconcilerStatusTest.java

@@ -11,7 +11,6 @@
 import io.fabric8.kubernetes.api.model.PodBuilder;
 import io.strimzi.api.kafka.model.kafka.Kafka;
 import io.strimzi.api.kafka.model.kafka.KafkaBuilder;
-import io.strimzi.api.kafka.model.kafka.KafkaMetadataState;
 import io.strimzi.api.kafka.model.kafka.KafkaResources;
 import io.strimzi.api.kafka.model.kafka.KafkaStatus;
 import io.strimzi.api.kafka.model.kafka.PersistentClaimStorageBuilder;
@@ -173,9 +172,6 @@ public void testKafkaReconcilerStatus(VertxTestContext context) {
             // Check kafka version
             assertThat(status.getKafkaVersion(), is(VERSIONS.defaultVersion().version()));
 
-            // Check Kafka metadata state
-            assertThat(status.getKafkaMetadataState(), is(KafkaMetadataState.KRaft));
-
             // Check model warning conditions
             assertThat(status.getConditions().size(), is(2));
             assertThat(status.getConditions().get(0).getType(), is("Warning"));

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaStatusTest.java

@@ -9,7 +9,6 @@
 import io.strimzi.api.kafka.model.kafka.Kafka;
 import io.strimzi.api.kafka.model.kafka.KafkaBuilder;
 import io.strimzi.api.kafka.model.kafka.KafkaList;
-import io.strimzi.api.kafka.model.kafka.KafkaMetadataState;
 import io.strimzi.api.kafka.model.kafka.KafkaStatus;
 import io.strimzi.api.kafka.model.kafka.listener.GenericKafkaListenerBuilder;
 import io.strimzi.api.kafka.model.kafka.listener.KafkaListenerType;
@@ -339,7 +338,6 @@ public void testStatusAfterFailedReconciliationWithPreviousSuccess(VertxTestCont
                     .withKafkaVersion("old-kafka")
                     .withOperatorLastSuccessfulVersion("old-operator")
                     .withKafkaMetadataVersion("old-metadata-version")
-                    .withKafkaMetadataState(KafkaMetadataState.KRaft)
                 .endStatus()
                 .build();
 
@@ -380,7 +378,6 @@ vertx, new PlatformFeaturesAvailability(false, kubernetesVersion),
             assertThat(status.getKafkaVersion(), is("old-kafka"));
             assertThat(status.getOperatorLastSuccessfulVersion(), is("old-operator"));
             assertThat(status.getKafkaMetadataVersion(), is("old-metadata-version"));
-            assertThat(status.getKafkaMetadataState(), is(KafkaMetadataState.KRaft));
 
             async.flag();
         })));