Enable multi version single step downgrade for KRaft based clusters

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

@@ -142,13 +142,14 @@ private Future<Void> detectToAndFromVersions(List<Pod> pods)    {
                 LOGGER.warnCr(reconciliation, "Kafka Pods exist, but do not contain the {} annotation to detect their version. Kafka upgrade cannot be detected.", ANNO_STRIMZI_IO_KAFKA_VERSION);
                 throw new KafkaUpgradeException("Kafka Pods exist, but do not contain the " + ANNO_STRIMZI_IO_KAFKA_VERSION + " annotation to detect their version. Kafka upgrade cannot be detected.");
             }
-        } else if (lowestKafkaVersion.equals(highestKafkaVersion)) {
-            // All brokers have the same version. We can use it as the current version.
-            versionFrom = versions.version(lowestKafkaVersion);
-            versionTo = versionFromCr;
         } else if (compareDottedVersions(highestKafkaVersion, versionFromCr.version()) > 0)    {
             // Highest Kafka version used by the brokers is higher than desired => suspected downgrade
-            versionFrom = versions.version(highestKafkaVersion);
+            try {
+                versionFrom = versions.version(highestKafkaVersion);
+            } catch (KafkaUpgradeException exception) {
+                // From version is unknown but thats ok for downgrade
+                versionFrom = new KafkaVersion(highestKafkaVersion, null, null, null, false, false, null);
+            }
             versionTo = versionFromCr;
         } else {
             // Highest Kafka version used by the brokers is equal or lower than desired => suspected upgrade

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

@@ -443,6 +443,31 @@ public void testDowngradeWithAllVersionAndMixedPods(VertxTestContext context) {
             async.flag();
         })));
     }
+
+    @Test
+    public void testDowngradeFromUnknownVersion(VertxTestContext context) {
+        String unknownVersion = getHigherVersionThanLatest();
+        KRaftVersionChangeCreator vcc = mockVersionChangeCreator(
+                mockKafka(VERSIONS.version(KafkaVersionTestUtils.LATEST_KAFKA_VERSION).version(), VERSIONS.version(KafkaVersionTestUtils.LATEST_KAFKA_VERSION).metadataVersion(), VERSIONS.version(KafkaVersionTestUtils.LATEST_KAFKA_VERSION).metadataVersion()),
+                mockRos(mockUniformPods(unknownVersion))
+        );
+
+        Checkpoint async = context.checkpoint();
+        vcc.reconcile().onComplete(context.succeeding(c -> context.verify(() -> {
+            assertThat(c.from().version(), is(unknownVersion));
+            assertThat(c.to(), is(VERSIONS.version(KafkaVersionTestUtils.LATEST_KAFKA_VERSION)));
+            assertThat(c.metadataVersion(), is(VERSIONS.version(KafkaVersionTestUtils.LATEST_KAFKA_VERSION).metadataVersion()));
+
+            async.flag();
+        })));
+    }
+
+    private String getHigherVersionThanLatest() {
+        String[] parsedVersion = KafkaVersionTestUtils.LATEST_KAFKA_VERSION.split("\\.");
+        int incrementedMinorVersion = Integer.parseInt(parsedVersion[1]) + 1;
+        parsedVersion[1] = Integer.toString(incrementedMinorVersion);
+        return String.join(".", parsedVersion);  
+    }
 
     @Test
     public void testDowngradeWithWrongCurrentMetadataVersion(VertxTestContext context) {

documentation/assemblies/upgrading/assembly-downgrade-cluster-operator.adoc

@@ -0,0 +1,26 @@
+// This assembly is included in the following assemblies:
+//
+// assembly-downgrade.adoc
+
+[id='assembly-downgrade-cluster-operator-{context}']
+= Downgrading the Cluster Operator
+
+//steps to downgrade the operators
+include::../../modules/upgrading/proc-downgrade-cluster-operator.adoc[leveloffset=+1]
+
+[id='con-downgrade-cluster-operator-unsupported-kafka-{context}']
+== Downgrading the Cluster Operator returns Kafka version error
+
+If you downgrade the Cluster Operator to a version that does not support the current version of Kafka you are using, you get an _unsupported Kafka version_ error.
+This error applies to all installation methods and means that you must downgrade Kafka to a supported Kafka version.
+Change the `spec.kafka.version` in the `Kafka` resource to the supported version.
+
+You can use `kubectl` to check for error messages like this in the `status` of the `Kafka` resource.
+
+.Checking the Kafka status for errors
+[source,shell, subs=+quotes]
+----
+kubectl get kafka <kafka_cluster_name> -n <namespace> -o jsonpath='{.status.conditions}'
+----
+
+Replace <kafka_cluster_name> with the name of your Kafka cluster and <namespace> with the Kubernetes namespace where the pod is running.

documentation/assemblies/upgrading/assembly-downgrade.adoc

@@ -17,8 +17,11 @@ WARNING: The following downgrade instructions are only suitable if you installed
 If you installed Strimzi using another method, like {OperatorHub}, downgrade may not be supported by that method unless specified in their documentation. 
 To ensure a successful downgrade process, it is essential to use a supported approach.
 
+//kafka downgrade paths
+include::../../modules/upgrading/con-downgrade-paths.adoc[leveloffset=+1]
+
 //steps to downgrade the operators
-include::../../modules/upgrading/proc-downgrade-cluster-operator.adoc[leveloffset=+1]
+include::assembly-downgrade-cluster-operator.adoc[leveloffset=+1]
 
 //steps to downgrade Kafka
 include::../../modules/upgrading/proc-downgrade-kafka-kraft.adoc[leveloffset=+1]

documentation/modules/upgrading/con-downgrade-paths.adoc

@@ -0,0 +1,43 @@
+// This assembly is included in the following assemblies:
+//
+// assembly-upgrade.adoc
+
+[id='con-downgrade-paths-{context}']
+= Strimzi downgrade paths
+
+[role="_abstract"]
+Two downgrade paths are available for Strimzi.
+
+Incremental downgrade::
+An incremental downgrade involves downgrading Strimzi from version {ProductVersion} to the previous minor version.
+
+Multi-version downgrade::
+A multi-version downgrade involves downgrading from version {ProductVersion} to an older version of Strimzi within a single upgrade, skipping one or more intermediate versions. 
+Downgrading to some older versions may not be sucessful, for example, due to configuration changes or extra steps for a particular Kafka version upgrade path performed by later versions that the target version is unable to undo. It is therefore necessary to validate your wanted downgrade path before attempting in a production environment. See table <<downgrade-version-limitations>> for details on specific paths.
+
+[id='con-downgrade-paths-kafka-versions-{context}']
+== Support for Kafka versions when downgrading
+
+When downgrading Strimzi, it is important to ensure compatibility with the Kafka version being used.
+
+Multi-version downgrades are possible even if the supported Kafka versions differ between the old and new versions. 
+However, if you attempt to downgrade to an older Strimzi version that does not support the current Kafka version, xref:con-downgrade-cluster-operator-unsupported-kafka-str[an error indicating that the Kafka version is not supported is generated]. 
+In this case, you must, if possible, downgrade the Kafka version as part of the Strimzi downgrade by changing the `spec.kafka.version` in the `Kafka` custom resource to the supported version for the target Strimzi version. If it is not possible to downgrade the Kafka version (for example in Strimzi version 0.45.0 or earlier), then the Strimzi downgrade path you are attempting is not supported. 
+
+The following table details known limitations for specific downgrade paths.
+
+.Known limitations in downgrade paths
+[[downgrade-version-limitations]]
+[cols="2,5",options="header"]
+|===
+
+| Target Strimzi Version
+| Limitation
+
+| <= 0.45.0
+| The current Kafka version must be supported by the target Strimzi version.
+
+It is only possible to downgrade from unsupported Kafka versions in Strimzi version 0.46.0 and higher.
+
+|===
+

documentation/modules/upgrading/proc-downgrade-cluster-operator.adoc

@@ -46,6 +46,18 @@ kubectl replace -f install/cluster-operator
 +
 Wait for the rolling updates to complete.
 
+. If the target operator version does not support the current Kafka version, an error message is returned.
++
+To resolve this, downgrade to a supported Kafka version if possible:
++
+--
+.. Edit the `Kafka` custom resource.
+.. Change the `spec.kafka.version` property to a supported Kafka version.
+--
++
+If no error message is returned, you can proceed to the next step and upgrade the Kafka version later.
+
+
 . Get the image for the Kafka pod to ensure the downgrade was successful:
 +
 [source,shell,subs="+quotes,attributes"]