Red Hat Summit Red Hat SummitSuporteConsoleDesenvolvedoresComeçar um testeContato

Este conteúdo não está disponível no idioma selecionado.

Chapter 9. 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.

9.1. Topic management modes
Copiar o link

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:

Bidirectional mode
Bidirectional mode requires ZooKeeper for cluster management. It is not compatible with using AMQ Streams in KRaft mode.
(Preview) Unidirectional mode
Unidirectional mode does not require ZooKeeper for cluster management. It is compatible with using AMQ Streams in KRaft mode.
Note

Unidirectional topic management is available as a preview. Unidirectional topic management is not enabled by default, so you must enable the UnidirectionalTopicOperator feature gate to be able to use it.

9.1.1. Bidirectional topic management
Copiar o link

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.

9.1.2. (Preview) Unidirectional topic management
Copiar o link

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.

9.2. Topic naming conventions
Copiar o link

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 (preview) 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.

9.3. Handling changes to topics
Copiar o link

How the Topic Operator handles changes to topics depends on the mode of topic management.

  • 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.
  • For unidirectional topic management (currently in preview), 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.

9.3.1. Topic store for bidirectional topic management
Copiar o link

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.

9.3.2. Migrating topic metadata from ZooKeeper to the topic store
Copiar o link

In previous releases of AMQ Streams, 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 AMQ Streams 2.5, 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 AMQ Streams 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-35-rhel8:2.5.2 --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.

9.3.4. Automatic creation of topics
Copiar o link

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 trying the unidirectional topic management preview, 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.

9.4. Configuring Kafka topics
Copiar o link

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 bidirectional and (preview) unidirectional 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

  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

9.5. Configuring topics for replication and number of partitions
Copiar o link

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 AMQ Streams 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.

9.6. (Preview) Managing KafkaTopic resources without impacting Kafka topics
Copiar o link

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.

Note

Unidirectional topic management is available as a preview. Unidirectional topic management is not enabled by default, so you must enable the UnidirectionalTopicOperator feature gate to be able to use it.

Prerequisites

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.

9.7. (Preview) Enabling topic management for existing Kafka topics
Copiar o link

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.

Note

Unidirectional topic management is available as a preview. Unidirectional topic management is not enabled by default, so you must enable the UnidirectionalTopicOperator feature gate to be able to use it.

Prerequisites

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).

9.8. (Preview) Deleting managed topics
Copiar o link

Unidirectional topic management supports the deletion of topics managed through the KafkaTopic resource with or without OpenShift finalizers. This is controlled 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.

Note

Unidirectional topic management is available as a preview. Unidirectional topic management is not enabled by default, so you must enable the UnidirectionalTopicOperator feature gate to be able to use it.

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
  finalizer:
    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
  finalizer:
    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 prevent related resources from being deleted. If the unidirectional Topic Operator is not running, it won’t be able to remove the metadata.finalizer. Consequently, an attempt to delete the namespace that contains the KafkaTopic resource won’t complete until either the operator is restarted, or the finalizer is otherwise removed (for example using oc edit).

Red Hat logoGithubredditYoutubeTwitter

Aprender

Experimente, compre e venda

Comunidades

Sobre a documentação da Red Hat

Ajudamos os usuários da Red Hat a inovar e atingir seus objetivos com nossos produtos e serviços com conteúdo em que podem confiar. Explore nossas atualizações recentes.

Tornando o open source mais inclusivo

A Red Hat está comprometida em substituir a linguagem problemática em nosso código, documentação e propriedades da web. Para mais detalhes veja o Blog da Red Hat.

Sobre a Red Hat

Fornecemos soluções robustas que facilitam o trabalho das empresas em plataformas e ambientes, desde o data center principal até a borda da rede.

Theme

© 2026 Red HatVoltar ao topo