// This module is included in the following assemblies:
//
// assembly-upgrade-zookeeper.adoc

[id='proc-upgrade-kafka-zookeeper-{context}']
= Upgrading ZooKeeper-based Kafka clusters

[role="_abstract"]
Upgrade a ZooKeeper-based Kafka cluster to a newer supported Kafka version and inter-broker protocol version.

.Prerequisites

* The Cluster Operator is up and running.
* Before you upgrade the Kafka cluster, check that the properties of the `Kafka` resource do _not_ contain configuration options that are not supported in the new Kafka version.

.Procedure

. Update the Kafka cluster configuration:
+
[source,shell,subs=+quotes]
----
kubectl edit kafka <kafka_configuration_file>
----

. If configured, check that the `inter.broker.protocol.version` and `log.message.format.version` properties are set to the _current_ version.
+
For example, the current version is {InterBrokerVersionLower} if upgrading from Kafka version {KafkaVersionLower} to {KafkaVersionHigher}:
+
[source,yaml,subs=attributes+]
----
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionLower}
    config:
      log.message.format.version: "{InterBrokerVersionLower}"
      inter.broker.protocol.version: "{InterBrokerVersionLower}"
      # ...
----
+
If `log.message.format.version` and `inter.broker.protocol.version` are not configured,
Strimzi automatically updates these versions to the current defaults after the update to the Kafka version in the next step.
+
NOTE: The value of `log.message.format.version` and `inter.broker.protocol.version` must be strings to prevent them from being interpreted as floating point numbers.

. Change the `Kafka.spec.kafka.version` to specify the new Kafka version; leave the `log.message.format.version` and `inter.broker.protocol.version` at the defaults for the _current_ Kafka version.
+
[NOTE]
====
Changing the `kafka.version` ensures that all brokers in the cluster are upgraded to start using the new broker binaries.
During this process, some brokers are using the old binaries while others have already upgraded to the new ones.
Leaving the `inter.broker.protocol.version` unchanged at the current setting ensures that the brokers can continue to communicate with each other throughout the upgrade.
====
+
For example, if upgrading from Kafka {KafkaVersionLower} to {KafkaVersionHigher}:
+
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionHigher} <1>
    config:
      log.message.format.version: "{InterBrokerVersionLower}" <2>
      inter.broker.protocol.version: "{InterBrokerVersionLower}" <3>
      # ...
----
<1> Kafka version is changed to the new version.
<2> Message format version is unchanged.
<3> Inter-broker protocol version is unchanged.
+
WARNING: You cannot downgrade Kafka if the `inter.broker.protocol.version` for the new Kafka version changes. The inter-broker protocol version determines the schemas used for persistent metadata stored by the broker, including messages written to `__consumer_offsets`. The downgraded cluster will not understand the messages.

. If the image for the Kafka cluster is defined in `Kafka.spec.kafka.image` of the `Kafka` custom resource, update the `image` to point to a container image with the new Kafka version.
+
See xref:con-versions-and-images-str[Kafka version and image mappings]

. Save and exit the editor, then wait for rolling updates to complete.
+
Check the progress of the rolling updates by watching the pod state transitions:
+
[source,shell,subs=+quotes]
----
kubectl get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'
----
+
The rolling updates ensure that each pod is using the broker binaries for the new version of Kafka.

. 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`.
.. For MirrorMaker 2, update `KafkaMirrorMaker2.spec.version`.
+
NOTE: If you are using custom images that are built manually, you must rebuild those images to ensure that they are up-to-date with the latest Strimzi base image. 
For example, if you xref:creating-new-image-from-base-str[created a container image from the base Kafka Connect image], update the Dockerfile to point to the latest base image and build configuration.

. If configured, update the Kafka resource to use the new `inter.broker.protocol.version` version. Otherwise, go to step 9.
+
For example, if upgrading to Kafka {KafkaVersionHigher}:
+
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionHigher}
    config:
      log.message.format.version: "{InterBrokerVersionLower}"
      inter.broker.protocol.version: "{InterBrokerVersionHigher}"
      # ...
----

. Wait for the Cluster Operator to update the cluster.

. If configured, update the Kafka resource to use the new `log.message.format.version` version. Otherwise, go to step 10.
+
For example, if upgrading to Kafka {KafkaVersionHigher}:
+
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  # ...
  kafka:
    version: {KafkaVersionHigher}
    config:
      log.message.format.version: "{InterBrokerVersionHigher}"
      inter.broker.protocol.version: "{InterBrokerVersionHigher}"
      # ...
----
+
IMPORTANT: From Kafka 3.0.0, when the `inter.broker.protocol.version` is set to `3.0` or higher, the `log.message.format.version` option is ignored and doesn't need to be set.

. Wait for the Cluster Operator to update the cluster.
+
Check the upgrade has completed successfully from the xref:con-upgrade-status-{context}[status of the `Kafka` resource].

.Upgrading Kafka client applications

Ensure all Kafka client applications are updated to use the new version of the client binaries as part of the upgrade process and verify their compatibility with the Kafka upgrade. 
If needed, coordinate with the team responsible for managing the client applications.

TIP: To check that a client is using the latest message format, use the `kafka.server:type=BrokerTopicMetrics,name={Produce|Fetch}MessageConversionsPerSec` metric. 
The metric shows `0` if the latest message format is being used.