Chapter 10. Using the Topic Operator to manage Kafka topics The KafkaTopic resource configures topics, including partition and replication factor settings. When you create, modify, or delete a topic using KafkaTopic , the Topic Operator ensures that these changes are reflected in the Kafka cluster. For more information on the KafkaTopic resource, see the KafkaTopic schema reference . 10.1. Topic management modes The KafkaTopic resource is responsible for managing a single topic within a Kafka cluster. The Topic Operator provides two modes for managing KafkaTopic resources and Kafka topics: Unidirectional mode (default) Unidirectional mode does not require ZooKeeper for cluster management. It is compatible with using Streams for Apache Kafka in KRaft mode. Bidirectional mode Bidirectional mode requires ZooKeeper for cluster management. It is not compatible with using Streams for Apache Kafka in KRaft mode. Note As the feature gate enabling the Topic Operator to run in unidirectional mode progresses to General Availability, bidirectional mode will be phased out. This transition is aimed at enhancing the user experience, particularly in supporting Kafka in KRaft mode. 10.1.1. Unidirectional topic management In unidirectional mode, the Topic Operator operates as follows: When a KafkaTopic is created, deleted, or changed, the Topic Operator performs the corresponding operation on the Kafka topic. If a topic is created, deleted, or modified directly within the Kafka cluster, without the presence of a corresponding KafkaTopic resource, the Topic Operator does not manage that topic. The Topic Operator will only manage Kafka topics associated with KafkaTopic resources and does not interfere with topics managed independently within the Kafka cluster. If a KafkaTopic does exist for a Kafka topic, any configuration changes made outside the resource are reverted. The Topic Operator can detect cases where where multiple KafkaTopic resources are attempting to manage a Kafka topic using the same .spec.topicName . Only the oldest resource is reconciled, while the other resources fail with a resource conflict error. 10.1.2. Bidirectional topic management In bidirectional mode, the Topic Operator operates as follows: When a KafkaTopic is created, deleted, or changed, the Topic Operator performs the corresponding operation on the Kafka topic. Similarly, when a topic is created, deleted, or changed within the Kafka cluster, the Topic Operator performs the corresponding operation on the KafkaTopic resource. Tip Try to stick to one method of managing topics, either through the KafkaTopic resources or directly in Kafka. Avoid routinely switching between both methods for a given topic. 10.2. Topic naming conventions A KafkaTopic resource includes a name for the topic and a label that identifies the name of the Kafka cluster it belongs to. Label identifying a Kafka cluster for topic handling apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: topic-name-1 labels: strimzi.io/cluster: my-cluster spec: topicName: topic-name-1 The label provides the cluster name of the Kafka resource. The Topic Operator uses the label as a mechanism for determining which KafkaTopic resources to manage. If the label does not match the Kafka cluster, the Topic Operator cannot see the KafkaTopic , and the topic is not created. Kafka and OpenShift have their own naming validation rules, and a Kafka topic name might not be a valid resource name in OpenShift. If possible, try and stick to a naming convention that works for both. Consider the following guidelines: Use topic names that reflect the nature of the topic Be concise and keep the name under 63 characters Use all lower case and hyphens Avoid special characters, spaces or symbols The KafkaTopic resource allows you to specify the Kafka topic name using the metadata.name field. However, if the desired Kafka topic name is not a valid OpenShift resource name, you can use the spec.topicName property to specify the actual name. The spec.topicName field is optional, and when it's absent, the Kafka topic name defaults to the metadata.name of the topic. When a topic is created, the topic name cannot be changed later. Example of supplying a valid Kafka topic name apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic-1 1 spec: topicName: My.Topic.1 2 # ... 1 A valid topic name that works in OpenShift. 2 A Kafka topic name that uses upper case and periods, which are invalid in OpenShift. If more than one KafkaTopic resource refers to the same Kafka topic, the resource that was created first is considered to be the one managing the topic. The status of the newer resources is updated to indicate a conflict, and their Ready status is changed to False . If a Kafka client application, such as Kafka Streams, automatically creates topics with invalid OpenShift resource names, the Topic Operator generates a valid metadata.name when used in bidirectional mode. It replaces invalid characters and appends a hash to the name. However, this behavior does not apply in unidirectional mode. Example of replacing an invalid topic name apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic---c55e57fe2546a33f9e603caf57165db4072e827e # ... Note For more information on the requirements for identifiers and names in a cluster, refer to the OpenShift documentation Object Names and IDs . 10.3. Handling changes to topics How the Topic Operator handles changes to topics depends on the mode of topic management . For unidirectional topic management, configuration changes only go in one direction: from the KafkaTopic resource to the Kafka topic. Any changes to a Kafka topic managed outside the KafkaTopic resource are reverted. For bidirectional topic management, configuration changes are synchronized between the Kafka topic and the KafkaTopic resource in both directions. Incompatible changes prioritize the Kafka configuration, and the KafkaTopic resource is adjusted accordingly. 10.3.1. Topic store for bidirectional topic management For bidirectional topic management, the Topic Operator is capable of handling changes to topics when there is no single source of truth. The KafkaTopic resource and the Kafka topic can undergo independent modifications, where real-time observation of changes may not always be feasible, particularly when the Topic Operator is not operational. To handle this, the Topic Operator maintains a topic store that stores topic configuration information about each topic. It compares the state of the Kafka cluster and OpenShift with the topic store to determine the necessary changes for synchronization. This evaluation takes place during startup and at regular intervals while the Topic Operator is active. For example, if the Topic Operator is inactive, and a new KafkaTopic named my-topic is created, upon restart, the Topic Operator recognizes the absence of my-topic in the topic store. It recognizes that the KafkaTopic was created after its last operation. Consequently, the Topic Operator generates the corresponding Kafka topic and saves the metadata in the topic store. The topic store enables the Topic Operator to manage situations where the topic configuration is altered in both Kafka topics and KafkaTopic resources, as long as the changes are compatible. When Kafka topic configuration is updated or changes are made to the KafkaTopic custom resource, the topic store is updated after reconciling with the Kafka cluster, as long as the changes are compatible. The topic store is based on the Kafka Streams key-value mechanism, which uses Kafka topics to persist the state. Topic metadata is cached in-memory and accessed locally within the Topic Operator. Updates from operations applied to the local in-memory cache are persisted to a backup topic store on disk. The topic store is continually synchronized with updates from Kafka topics or OpenShift KafkaTopic custom resources. Operations are handled rapidly with the topic store set up this way, but should the in-memory cache crash it is automatically repopulated from the persistent storage. Internal topics support the handling of topic metadata in the topic store. __strimzi_store_topic Input topic for storing the topic metadata __strimzi-topic-operator-kstreams-topic-store-changelog Retains a log of compacted topic store values Warning Do not delete these topics, as they are essential to the running of the Topic Operator. 10.3.2. Migrating topic metadata from ZooKeeper to the topic store In releases of Streams for Apache Kafka, topic metadata was stored in ZooKeeper. The topic store removes this requirement, bringing the metadata into the Kafka cluster, and under the control of the Topic Operator. When upgrading to Streams for Apache Kafka 2.7, the transition to Topic Operator control of the topic store is seamless. Metadata is found and migrated from ZooKeeper, and the old store is deleted. 10.3.3. Downgrading to a Streams for Apache Kafka version that uses ZooKeeper to store topic metadata If you are reverting back to a version of Streams for Apache Kafka earlier than 1.7, which uses ZooKeeper for the storage of topic metadata, you still downgrade your Cluster Operator to the version, then downgrade Kafka brokers and client applications to the Kafka version as standard. However, you must also delete the topics that were created for the topic store using a kafka-topics command, specifying the bootstrap address of the Kafka cluster. For example: oc run kafka-admin -ti --image=registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 --rm=true --restart=Never -- ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi-topic-operator-kstreams-topic-store-changelog --delete && ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi_store_topic --delete The command must correspond to the type of listener and authentication used to access the Kafka cluster. The Topic Operator will reconstruct the ZooKeeper topic metadata from the state of the topics in Kafka. 10.3.4. Automatic creation of topics Applications can trigger the automatic creation of topics in the Kafka cluster. By default, the Kafka broker configuration auto.create.topics.enable is set to true , allowing the broker to create topics automatically when an application attempts to produce or consume from a non-existing topic. Applications might also use the Kafka AdminClient to automatically create topics. When an application is deployed along with its KafkaTopic resources, it is possible that automatic topic creation in the cluster happens before the Topic Operator can react to the KafkaTopic . For bidirectional topic management, the Topic Operator synchronizes the changes between the topics and KafkaTopic resources. If you are using unidirectional topic management, this can mean that the topics created for an application deployment are initially created with default topic configuration. If the Topic Operator attempts to reconfigure the topics based on KafkaTopic resource specifications included with the application deployment, the operation might fail because the required change to the configuration is not allowed. For example, if the change means lowering the number of topic partitions. For this reason, it is recommended to disable auto.create.topics.enable in the Kafka cluster configuration when using unidirectional topic management. 10.4. Configuring Kafka topics Use the properties of the KafkaTopic resource to configure Kafka topics. Changes made to topic configuration in the KafkaTopic are propagated to Kafka. You can use oc apply to create or modify topics, and oc delete to delete existing topics. For example: oc apply -f <topic_config_file> oc delete KafkaTopic <topic_name> To be able to delete topics, delete.topic.enable must be set to true (default) in the spec.kafka.config of the Kafka resource. This procedure shows how to create a topic with 10 partitions and 2 replicas. Note The procedure is the same for the unidirectional and bidirectional modes of topic management. Before you begin The KafkaTopic resource does not allow the following changes: Renaming the topic defined in spec.topicName . A mismatch between spec.topicName and status.topicName will be detected. Decreasing the number of partitions using spec.partitions (not supported by Kafka). Modifying the number of replicas specified in spec.replicas . Warning Increasing spec.partitions for topics with keys will alter the partitioning of records, which can cause issues, especially when the topic uses semantic partitioning. Prerequisites A running Kafka cluster configured with a Kafka broker listener using mTLS authentication and TLS encryption. A running Topic Operator (typically deployed with the Entity Operator). For deleting a topic, delete.topic.enable=true (default) in the spec.kafka.config of the Kafka resource. Procedure 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 . Create the KafkaTopic resource in OpenShift. oc apply -f <topic_config_file> 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 . 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 10.5. Configuring topics for replication and number of partitions The recommended configuration for topics managed by the Topic Operator is a topic replication factor of 3, and a minimum of 2 in-sync replicas. apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic labels: strimzi.io/cluster: my-cluster spec: partitions: 10 1 replicas: 3 2 config: min.insync.replicas: 2 3 #... 1 The number of partitions for the topic. 2 The number of replica topic partitions. Currently, this cannot be changed in the KafkaTopic resource, but it can be changed using the kafka-reassign-partitions.sh tool. 3 The minimum number of replica partitions that a message must be successfully written to, or an exception is raised. Note In-sync replicas are used in conjunction with the acks configuration for producer applications. The acks configuration determines the number of follower partitions a message must be replicated to before the message is acknowledged as successfully received. The bidirectional Topic Operator runs with acks=all for its internal topics whereby messages must be acknowledged by all in-sync replicas. When scaling Kafka clusters by adding or removing brokers, replication factor configuration is not changed and replicas are not reassigned automatically. However, you can use the kafka-reassign-partitions.sh tool to change the replication factor, and manually reassign replicas to brokers. Alternatively, though the integration of Cruise Control for Streams for Apache Kafka cannot change the replication factor for topics, the optimization proposals it generates for rebalancing Kafka include commands that transfer partition replicas and change partition leadership. Additional resources Downgrading Streams for Apache Kafka Section 20.1, "Partition reassignment tool overview" Chapter 19, Rebalancing clusters using Cruise Control 10.6. Managing KafkaTopic resources without impacting Kafka topics This procedure describes how to convert Kafka topics that are currently managed through the KafkaTopic resource into non-managed topics. This capability can be useful in various scenarios. For instance, you might want to update the metadata.name of a KafkaTopic resource. You can only do that by deleting the original KafkaTopic resource and recreating a new one. By annotating a KafkaTopic resource with strimzi.io/managed=false , you indicate that the Topic Operator should no longer manage that particular topic. This allows you to retain the Kafka topic while making changes to the resource's configuration or other administrative tasks. You can perform this task if you are using unidirectional topic management. Prerequisites The Cluster Operator must be deployed. Procedure 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. 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). 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: Delete the original KafkTopic resource: oc delete kafkatopic <kafka_topic_name> 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 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. 10.7. Enabling topic management for existing Kafka topics This procedure describes how to enable topic management for topics that are not currently managed through the KafkaTopic resource. You do this by creating a matching KafkaTopic resource. You can perform this task if you are using unidirectional topic management. Prerequisites The Cluster Operator must be deployed. Procedure 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. Apply the KafkaTopic resource: oc apply -f <topic_configuration_file> 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. 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). 10.8. Deleting managed topics Unidirectional topic management supports the deletion of topics managed through the KafkaTopic resource with or without OpenShift finalizers. This is determined by the STRIMZI_USE_FINALIZERS Topic Operator environment variable. By default, this is set to true , though it can be set to false in the Topic Operator env configuration if you do not want the Topic Operator to add finalizers. Finalizers ensure orderly and controlled deletion of KafkaTopic resources. A finalizer for the Topic Operator is added to the metadata of the KafkaTopic resource: Finalizer to control topic deletion apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: generation: 1 name: my-topic-1 finalizers: - strimzi.io/topic-operator labels: strimzi.io/cluster: my-cluster In this example, the finalizer is added for topic my-topic-1 . The finalizer prevents the topic from being fully deleted until the finalization process is complete. If you then delete the topic using oc delete kafkatopic my-topic-1 , a timestamp is added to the metadata: Finalizer timestamp on deletion apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: generation: 1 name: my-topic-1 finalizers: - strimzi.io/topic-operator labels: strimzi.io/cluster: my-cluster deletionTimestamp: 20230301T000000.000 The resource is still present. If the deletion fails, it is shown in the status of the resource. When the finalization tasks are successfully executed, the finalizer is removed from the metadata, and the resource is fully deleted. Finalizers also serve to prevent related resources from being deleted. If the unidirectional Topic Operator is not running, it won't be able to remove its finalizer from the metadata.finalizers . And any attempt to directly delete the KafkaTopic resources or the namespace will fail or timeout, leaving the namespace in a stuck terminating state. If this happens, you can bypass the finalization process by removing the finalizers on topics . 10.9. Switching between Topic Operator modes You can switch between topic management modes when upgrading or downgrading Streams for Apache Kafka, or when using the same version of Streams for Apache Kafka, as long as the mode is supported for that version. Switching from bidirectional to unidirectional topic management mode Enable the UnidirectionalTopicOperator feature gate . The Cluster Operator deploys the Entity Operator with the Topic Operator in unidirectional topic management mode. 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 USD(oc get kt -n <namespace_name> -o name | grep strimzi-store-topic) \ && oc delete USD(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 . 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. Discontinue management of the topics: oc annotate USD(oc get kt -n <namespace_name> -o name | grep consumer-offsets) strimzi.io/managed="false" \ && oc annotate USD(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 . When their management is discontinued, delete the KafkaTopic resources (without deleting the topics inside Kafka): oc delete USD(oc get kt -n <namespace_name> -o name | grep consumer-offsets) \ && oc delete USD(oc get kt -n <namespace_name> -o name | grep transaction-state) Switching from unidirectional to bidirectional topic management mode 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. 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. After switching between between Topic Operator modes, try creating a topic to make sure the operator is running correctly. For more information, see Section 10.4, "Configuring Kafka topics" . 10.10. Removing finalizers on topics If the unidirectional Topic Operator is not running, and you want to bypass the finalization process when deleting managed topics, you have to remove the finalizers. You can do this manually by editing the resources directly or by using a command. To remove finalizers on all topics, use the following command: Removing finalizers on topics oc get kt -o=json | jq '.items[].metadata.finalizers = null' | oc apply -f - The command uses the jq command line JSON parser tool to modify the KafkaTopic ( kt ) resources by setting the finalizers to null . You can also use the command for a specific topic: Removing a finalizer on a specific topic oc get kt <topic_name> -o=json | jq '.metadata.finalizers = null' | oc apply -f - After running the command, you can go ahead and delete the topics. Alternatively, if the topics were already being deleted but were blocked due to outstanding finalizers then their deletion should complete. Warning Be careful when removing finalizers, as any cleanup operations associated with the finalization process are not performed if the Topic Operator is not running. For example, if you remove the finalizer from a KafkaTopic resource and subsequently delete the resource, the related Kafka topic won't be deleted. 10.11. Considerations when disabling topic deletion When the delete.topic.enable configuration in Kafka is set to false , topics cannot be deleted. This might be required in certain scenarios, but it introduces a consideration when using the Topic Operator in unidirectional mode. As topics cannot be deleted, finalizers added to the metadata of a KafkaTopic resource to control topic deletion are never removed by the Topic Operator (though they can be removed manually ). Similarly, any Custom Resource Definitions (CRDs) or namespaces associated with topics cannot be deleted. Before configuring delete.topic.enable=false , assess these implications to ensure it aligns with your specific requirements. Note To avoid using finalizers, you can set the STRIMZI_USE_FINALIZERS Topic Operator environment variable to false . 10.12. Tuning request batches for topic operations In unidirectional mode, the Topic Operator uses the request batching capabilities of the Kafka Admin API for operations on topic resources. You can fine-tune the batching mechanism using the following operator configuration properties: STRIMZI_MAX_QUEUE_SIZE to set the maximum size of the topic event queue. The default value is 1024. STRIMZI_MAX_BATCH_SIZE to set the maximum number of topic events allowed in a single batch. The default value is 100. MAX_BATCH_LINGER_MS to specify the maximum time to wait for a batch to accumulate items before processing. The default is 100 milliseconds. If the maximum size of the request batching queue is exceeded, the Topic Operator shuts down and is restarted. To prevent frequent restarts, consider adjusting the STRIMZI_MAX_QUEUE_SIZE property to accommodate the typical load.

Commands:
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: topic-name-1 labels: strimzi.io/cluster: my-cluster spec: topicName: topic-name-1
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic-1 1 spec: topicName: My.Topic.1 2 #
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic---c55e57fe2546a33f9e603caf57165db4072e827e #
run kafka-admin -ti --image=registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 --rm=true --restart=Never -- ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi-topic-operator-kstreams-topic-store-changelog --delete && ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi_store_topic --delete
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic-1 labels: strimzi.io/cluster: my-cluster spec: partitions: 10 replicas: 2
apply -f <topic_config_file>
get kafkatopics -o wide -w -n <namespace>
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
get kafkatopics my-topic-2 -o yaml
status: conditions: - lastTransitionTime: "2022-06-13T10:14:43.351550Z" message: Number of partitions cannot be decreased reason: PartitionDecreaseException status: "True" type: NotReady
get kafkatopics my-topic-2 -o wide -w -n <namespace>
NAME CLUSTER PARTITIONS REPLICATION FACTOR READY my-topic-2 my-cluster 10 3 True
get kafkatopics my-topic-2 -o yaml
status: conditions: - lastTransitionTime: '2022-06-13T10:15:03.761084Z' status: 'True' type: Ready
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic labels: strimzi.io/cluster: my-cluster spec: partitions: 10 1 replicas: 3 2 config: min.insync.replicas: 2 3 #
annotate kafkatopic my-topic-1 strimzi.io/managed="false"
get kafkatopics my-topic-1 -o yaml
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
delete kafkatopic <kafka_topic_name>
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name: my-topic-1 labels: strimzi.io/cluster: my-cluster spec: partitions: 10 replicas: 2
apply -f <topic_configuration_file>
get kafkatopics my-topic-1 -o yaml
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
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: generation: 1 name: my-topic-1 finalizers: - strimzi.io/topic-operator labels: strimzi.io/cluster: my-cluster
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: generation: 1 name: my-topic-1 finalizers: - strimzi.io/topic-operator labels: strimzi.io/cluster: my-cluster deletionTimestamp: 20230301T000000.000
delete USD(oc get kt -n <namespace_name> -o name | grep strimzi-store-topic) && oc delete USD(oc get kt -n <namespace_name> -o name | grep strimzi-topic-operator)
annotate USD(oc get kt -n <namespace_name> -o name | grep consumer-offsets) strimzi.io/managed="false" && oc annotate USD(oc get kt -n <namespace_name> -o name | grep transaction-state) strimzi.io/managed="false"
delete USD(oc get kt -n <namespace_name> -o name | grep consumer-offsets) && oc delete USD(oc get kt -n <namespace_name> -o name | grep transaction-state)
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: # entityOperator: topicOperator: {} userOperator: {} template: topicOperatorContainer: env: - name: STRIMZI_USE_FINALIZERS value: "false"
apiVersion: apps/v1 kind: Deployment metadata: name: strimzi-topic-operator spec: template: spec: containers: - name: STRIMZI_USE_FINALIZERS value: "false"
get kt -o=json | jq '.items[].metadata.finalizers = null' | oc apply -f -
get kt <topic_name> -o=json | jq '.metadata.finalizers = null' | oc apply -f -