Chapter 10. Using the Topic Operator to manage Kafka topics

Procedure

1. Configure the KafkaTopic resource.

   Example Kafka topic configuration

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     name: my-topic-1
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     partitions: 10
     replicas: 2

   Tip

   When modifying a topic, you can get the current version of the resource using oc get kafkatopic my-topic-1 -o yaml.

2. Create the KafkaTopic resource in OpenShift.

   oc apply -f <topic_config_file>

3. Wait for the ready status of the topic to change to True:

   oc get kafkatopics -o wide -w -n <namespace>

   Kafka topic status

   NAME         CLUSTER     PARTITIONS  REPLICATION FACTOR READY
   my-topic-1   my-cluster  10          3                  True
   my-topic-2   my-cluster  10          3
   my-topic-3   my-cluster  10          3                  True

   Topic creation is successful when the READY output shows True.

4. If the READY column stays blank, get more details on the status from the resource YAML or from the Topic Operator logs.

   Status messages provide details on the reason for the current status.

   oc get kafkatopics my-topic-2 -o yaml

   Details on a topic with a NotReady status

   # ...
   status:
     conditions:
     - lastTransitionTime: "2022-06-13T10:14:43.351550Z"
       message: Number of partitions cannot be decreased
       reason: PartitionDecreaseException
       status: "True"
       type: NotReady

   In this example, the reason the topic is not ready is because the original number of partitions was reduced in the KafkaTopic configuration. Kafka does not support this.

   After resetting the topic configuration, the status shows the topic is ready.

   oc get kafkatopics my-topic-2 -o wide -w -n <namespace>

   Status update of the topic

   NAME         CLUSTER     PARTITIONS  REPLICATION FACTOR READY
   my-topic-2   my-cluster  10          3                  True

   Fetching the details shows no messages

   oc get kafkatopics my-topic-2 -o yaml

   Details on a topic with a READY status

   # ...
   status:
     conditions:
     - lastTransitionTime: '2022-06-13T10:15:03.761084Z'
       status: 'True'
       type: Ready

Procedure

1. Annotate the KafkaTopic resource in OpenShift, setting strimzi.io/managed to false:

   oc annotate kafkatopic my-topic-1 strimzi.io/managed="false"

   Specify the metadata.name of the topic in your KafkaTopic resource, which is my-topic-1 in this example.

2. Check the status of the KafkaTopic resource to make sure the request was successful:

   oc get kafkatopics my-topic-1 -o yaml

   Example topic with a Ready status

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     generation: 124
     name: my-topic-1
     finalizer:
       strimzi.io/topic-operator
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     partitions: 10
     replicas: 2
   
   # ...
   status:
     observedGeneration: 124 1
     topicName: my-topic-1
     conditions:
     - type: Ready
       status: True
       lastTransitionTime: 20230301T103000Z

   1

   Successful reconciliation of the resource means the topic is no longer managed.

   The value of metadata.generation (the current version of the deployment) must match status.observedGeneration (the latest reconciliation of the resource).

3. You can now make changes to the KafkaTopic resource without it affecting the Kafka topic it was managing.

   For example, to change the metadata.name, do as follows:

   1. Delete the original KafkTopic resource:

      oc delete kafkatopic <kafka_topic_name>

   2. Recreate the KafkTopic resource with a different metadata.name, but use spec.topicName to refer to the same topic that was managed by the original
4. If you haven’t deleted the original KafkaTopic resource, and you wish to resume management of the Kafka topic again, set the strimzi.io/managed annotation to true or remove the annotation.

Procedure

1. Create a KafkaTopic resource with a metadata.name that is the same as the Kafka topic.

   Or use spec.topicName if the name of the topic in Kafka would not be a legal OpenShift resource name.

   Example Kafka topic configuration

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     name: my-topic-1
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     partitions: 10
     replicas: 2

   In this example, the Kafka topic is named my-topic-1.

   The Topic Operator checks whether the topic is managed by another KafkaTopic resource. If it is, the older resource takes precedence and a resource conflict error is returned in the status of the new resource.

2. Apply the KafkaTopic resource:

   oc apply -f <topic_configuration_file>

3. Wait for the operator to update the topic in Kafka.

   The operator updates the Kafka topic with the spec of the KafkaTopic that has the same name.

4. Check the status of the KafkaTopic resource to make sure the request was successful:

   oc get kafkatopics my-topic-1 -o yaml

   Example topic with a Ready status

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     generation: 1
     name: my-topic-1
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     partitions: 10
     replicas: 2
   # ...
   status:
     observedGeneration: 1 1
     topicName: my-topic-1
     conditions:
     - type: Ready
       status: True
       lastTransitionTime: 20230301T103000Z

   1

   Successful reconciliation of the resource means the topic is now managed.

   The value of metadata.generation (the current version of the deployment) must match status.observedGeneration (the latest reconciliation of the resource).

Switching from bidirectional to unidirectional topic management mode

1. Enable the UnidirectionalTopicOperator feature gate.

   The Cluster Operator deploys the Entity Operator with the Topic Operator in unidirectional topic management mode.

2. The internal topics that support the Topic Operator running in bidirectional topic management mode are no longer required, so you can delete the KafkaTopic resources to manage them:

   oc delete $(oc get kt -n <namespace_name> -o name | grep strimzi-store-topic) \
     && oc delete $(oc get kt -n <namespace_name> -o name | grep strimzi-topic-operator)

   This command deletes the internal topics, which have names starting strimzi-store-topic and strimzi-topic-operator.

3. The internal topics for storing consumer offsets and transaction states must be retained in Kafka. So, you must first discontinue their management by the Topic Operator before deleting the KafkaTopic resources.

   1. Discontinue management of the topics:

      oc annotate $(oc get kt -n <namespace_name> -o name | grep consumer-offsets) strimzi.io/managed="false" \
        && oc annotate $(oc get kt -n <namespace_name> -o name | grep transaction-state) strimzi.io/managed="false"

      By annotating the KafkaTopic resources with strimzi.io/managed="false", you indicate that the Topic Operator should no longer manage those topics. In this case, we are adding the annotation to resources for managing the internal topics with names starting consumer-offsets and transaction-state.

   2. When their management is discontinued, delete the KafkaTopic resources (without deleting the topics inside Kafka):

      oc delete $(oc get kt -n <namespace_name> -o name | grep consumer-offsets) \
        && oc delete $(oc get kt -n <namespace_name> -o name | grep transaction-state)

Switching from unidirectional to bidirectional topic management mode

1. Disable the UnidirectionalTopicOperator feature gate.

   The Cluster Operator deploys the Entity Operator with the Topic Operator in bidirectional topic management mode.

   The internal topics required by the Topic Operator running in bidirectional topic management mode are created.

2. Check whether finalizers are being used to control topic deletion. If KafkaTopic resources are using finalizers, ensure that you do one of the following after making the switch:

   • Remove all finalizers from topics.

   • Disable the finalizers by setting the STRIMZI_USE_FINALIZERS environment variable to false in the Topic Operator env configuration.

     Use the same configuration for a Topic Operator running in a Streams for Apache Kafka-managed cluster or as a standalone deployment.

     Disabling topic finalizers in a Streams for Apache Kafka-managed cluster

     apiVersion: kafka.strimzi.io/v1beta2
     kind: Kafka
     metadata:
       name: my-cluster
     spec:
       # ...
       entityOperator:
         topicOperator: {}
         userOperator: {}
         template:
           topicOperatorContainer:
             env:
               - name: STRIMZI_USE_FINALIZERS
                 value: "false"
     # ...

     Disabling topic finalizers in a standalone deployment

     apiVersion: apps/v1
     kind: Deployment
     metadata:
       name: strimzi-topic-operator
     spec:
       template:
         spec:
           containers:
             - name: STRIMZI_USE_FINALIZERS
               value: "false"
     # ...

     The Topic Operator does not use finalizers in bidirectional mode. If they are retained after making the switch from unidirectional mode, you won’t be able to delete KafkaTopic and related resources.