// This module is included in the following assemblies:
//
// upgrading/assembly_upgrade-kafka-versions.adoc

[id='proc-upgrading-brokers-newer-kafka-{context}']

= Upgrading Kafka brokers and client applications

This procedure describes how to upgrade a Strimzi Kafka cluster to a higher version of Kafka.

.Prerequisites

For the `Kafka` resource to be upgraded, 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 that you are upgrading to.
* Whether the `log.message.format.version` for the current Kafka version needs to be updated for the new version.
+
xref:ref-kafka-versions-{context}[Consult the Kafka versions table].

.Procedure

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

.. If the `log.message.format.version` of the current Kafka version is the same as that of the new Kafka version, proceed to the next step.
+
Otherwise, ensure that `Kafka.spec.kafka.config` has the `log.message.format.version` configured to the default for the _current_ version.
+
For example, if upgrading from Kafka {KafkaVersionLower}:
+
[source,yaml,subs=attributes+]
----
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionLower}
    config:
      log.message.format.version: "{LogMsgVersLower}"
      # ...
----
+
If the `log.message.format.version` is unset, set it to the current version.
+
NOTE: The value of `log.message.format.version` must be a string to prevent it from being interpreted as a floating point number.

.. Change the `Kafka.spec.kafka.version` to specify the new version (leaving the `log.message.format.version` as the current version).
+
For example, if upgrading from Kafka {KafkaVersionLower} to {KafkaVersionHigher}:
+
[source,yaml,subs=attributes+]
----
apiVersion: v1alpha1
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionHigher} <1>
    config:
      log.message.format.version: "{LogMsgVersLower}" <2>
      # ...
----
<1> This is changed to the new version
<2> This remains at the current version

.. 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[Kafka version and image mappings]

.. If the Kafka listener configuration does not use the array format of the `GenericKafkaListener` schema,
update `Kafka.spec.kafka.listeners`.
+
See xref:con-upgrade-listeners-str[Updating listener configuration]

. Save and exit the editor, then wait for rolling updates to complete.
+
NOTE: Additional rolling updates occur if the new version of Kafka has a new ZooKeeper version.
+
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 upgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"
----
+
[source,shell,subs=+quotes]
----
kubectl get po -w
----
+
====
If the current and new 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 upgrade from _<from-version>_ to _<to-version>_, phase 2 of 2 completed
----
Alternatively, if the current and new versions of Kafka have the same interbroker protocol version, check for:

[source,shell,subs=+quotes]
----
Reconciliation #_<num>_(watch) Kafka(_<namespace>_/_<name>_): Kafka version upgrade from _<from-version>_ to _<to-version>_, phase 1 of 1 completed
----
====
+
The rolling updates:
+
* Ensure each pod is using the broker binaries for the new version of Kafka
* Configure the brokers to send messages using the interbroker protocol of the new version of Kafka
+
NOTE: Clients are still using the old version, so brokers will convert messages to the old version before sending them to the clients. To minimize this additional load, update the clients as quickly as possible.

. Depending on your chosen strategy for upgrading clients, upgrade all client applications to use the new version of the client binaries.
+
See xref:con-strategies-for-upgrading-clients-str[Strategies for upgrading clients]
+
WARNING: You cannot downgrade after completing this step. If you need to revert the update at this point, follow the procedure xref:proc-downgrading-brokers-older-kafka-{context}[].

+
If required, set the version property for Kafka Connect and MirrorMaker as the new version of Kafka:
+
.. For Kafka Connect, update `KafkaConnect.spec.version`
.. For MirrorMaker, update `KafkaMirrorMaker.spec.version`

. If the `log.message.format.version` identified in step 1 is the same as the new version proceed to the next step.
+
Otherwise change the `log.message.format.version` in `Kafka.spec.kafka.config` to the default version for the new version of Kafka now being used.
+
For example, if upgrading to {KafkaVersionHigher}:
+
[source,yaml,subs=attributes+]
----
apiVersion: v1alpha1
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionHigher}
    config:
      log.message.format.version: "{LogMsgVersHigher}"
      # ...
----

. Wait for the Cluster Operator to update the cluster.
+
The Kafka cluster and clients are now using the new Kafka version.
