// This module is included in the following assemblies:
//
// assembly-downgrading-kafka-versions.adoc

[id='proc-downgrading-brokers-older-kafka-{context}']

= Downgrading Kafka brokers and client applications

This procedure describes how you can downgrade a {ProductName} Kafka cluster to a lower (previous) version of Kafka, such as downgrading from {KafkaVersionHigher} to {KafkaVersionLower}.

[IMPORTANT]
====
Downgrading is _not possible_ if the new version has ever used a `log.message.format.version` that is not supported by the previous version, including when the default value for `log.message.format.version` is used. For example, this resource can be downgraded to Kafka version {KafkaVersionLower} because the `log.message.format.version` has not been changed:

[source,yaml,subs=attributes+]
----
apiVersion: v1alpha1
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionHigher}
    config:
      log.message.format.version: "{LogMsgVersLower}"
      # ...
----

The downgrade would not be possible if the `log.message.format.version` was set at `"{LogMsgVersHigher}"` or a value was absent (so that the parameter took the default value for a {KafkaVersionHigher} broker of {LogMsgVersHigher}).
====

.Prerequisites

For the `Kafka` resource to be downgraded, check:

* The Cluster Operator, which supports both versions of Kafka, is up and running.
* The `Kafka.spec.kafka.config` does not contain options that are not supported in the version of Kafka you are downgrading to.
* The `Kafka.spec.kafka.config` has a `log.message.format.version` that is supported by the version being downgraded to.

.Procedure

. Update the Kafka cluster configuration in an editor, as required:
+
Use `kubectl edit`:
+
[source,shell,subs=+quotes]
kubectl edit kafka _my-cluster_

.. Change the `Kafka.spec.kafka.version` to specify the previous version.
+
For example, if downgrading from Kafka {KafkaVersionHigher} to {KafkaVersionLower}:
+
[source,yaml,subs=attributes+]
----
apiVersion: v1alpha1
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionLower} <1>
    config:
      log.message.format.version: "{LogMsgVersLower}" <2>
      # ...
----
<1> This is changed to the previous version
<2> This is unchanged
+
NOTE: You must format the value of `log.message.format.version` as a string to prevent it from being interpreted as a floating point number.

.. If the image for the Kafka version is different from the image defined in `STRIMZI_KAFKA_IMAGES` for the Cluster Operator, update `Kafka.spec.kafka.image`.
+
See xref:con-versions-and-images-str[]

. Save and exit the editor, then wait for rolling updates to complete.
+
Check the update in the logs or by watching the pod state transitions:
+
[source,shell,subs=+quotes]
----
kubectl logs -f _<cluster-operator-pod-name>_ | grep -E "Kafka version downgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"
----
+
[source,shell,subs=+quotes]
----
kubectl get po -w
----
+
====
If the previous and current versions of Kafka have different interbroker protocol versions, check the Cluster Operator logs for an `INFO` level message:

[source,shell,subs=+quotes]
----
Reconciliation #_<num>_(watch) Kafka(_<namespace>_/_<name>_): Kafka version downgrade from _<from-version>_ to _<to-version>_, phase 2 of 2 completed
----
Alternatively, if the previous and current versions of Kafka have the same interbroker protocol version, check for:

[source,shell,subs=+quotes]
----
Reconciliation #_<num>_(watch) Kafka(_<namespace>_/_<name>_): Kafka version downgrade from _<from-version>_ to _<to-version>_, phase 1 of 1 completed
----
====
+
. Downgrade all client applications (consumers) to use the previous version of the client binaries.
+
The Kafka cluster and clients are now using the previous Kafka version.
