10.3. Handling changes to topics

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.

In previous 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.

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 previous version, then downgrade Kafka brokers and client applications to the previous 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.

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.