Topics
A KafkaTopic declares the desired state of a topic in an Apache Kafka cluster: its name, partitions, replication factor and configs. The agent of the cluster creates the topic and keeps it in this state. This requires management access to be enabled on the KafkaCluster.
The KafkaTopic is the source of truth for its topic: changes made to the topic outside of the KafkaTopic, for example with kafka-configs.sh, are reverted.
The agent checks every topic at least every five minutes.
| With management access enabled, anyone who can create KafkaTopics in the namespace of the KafkaCluster controls all topics of that cluster, including existing ones and their deletion. Kafka’s own authorization only applies to the agent, not to the user of a KafkaTopic. |
Other operators, such as Strimzi, also provide a resource of the kind KafkaTopic.
If both are installed, use the full resource name with kubectl, for example kubectl get kafkatopics.kafka.stackable.tech.
|
Configuration
---
apiVersion: kafka.stackable.tech/v1alpha1
kind: KafkaTopic
metadata:
name: orders
spec:
clusterRef:
name: simple-kafka (1)
topicName: shop.orders (2)
partitions: 6 (3)
replicationFactor: 3 (4)
configOverrides: (5)
retention.ms: "604800000"
cleanup.policy: delete
deletionPolicy: Retain (6)| 1 | The KafkaCluster in the same namespace that manages the topic. It cannot be changed. |
| 2 | The name of the topic in Kafka. Defaults to metadata.name. It cannot be changed. |
| 3 | The number of partitions. Defaults to num.partitions of the KRaft controllers when the topic is created. |
| 4 | The number of replicas of each partition. Defaults to default.replication.factor of the KRaft controllers when the topic is created. |
| 5 | The configs of the topic, see Configs. |
| 6 | What happens to the topic when the KafkaTopic is deleted, see Deletion. |
Topic name
A topic name consists of up to 249 characters out of a-z, A-Z, 0-9, ., _ and -, and cannot be . or ...
Kafka allows names that are not valid Kubernetes object names, for example with uppercase letters or underscores.
Such topics need topicName.
The prefix __ is reserved for the internal topics of Kafka, such as __consumer_offsets, and is rejected.
The topic name cannot be changed, and topicName can neither be added nor removed later.
Kafka cannot rename topics, and a KafkaTopic with another name creates a new topic.
Partitions and replication factor
The number of partitions can only be increased, because Kafka cannot remove partitions.
| Increasing the partitions changes which partition a key is assigned to. Consumers that rely on the order of messages with the same key are affected. |
Changing the replication factor of an existing topic is not supported, because Kafka requires an explicit assignment of the replicas to brokers.
Use kafka-reassign-partitions.sh or Cruise Control for this.
An unsupported change is reported in the Synced condition, and all other changes are still applied.
The agent creates a topic as soon as it is able to reach a broker.
This could be while the KafkaCluster is still starting and only a subset of the brokers being online.
Kafka assigns the partitions to the brokers that are registered at that time, and the agent never reassigns partitions.
We recommend to delay creating KafkaTopics until the KafkaCluster is ready.
However, if for some reason the topic was created earlier (with only a subset of brokers), partitions of a topic can be spread with kafka-reassign-partitions.sh as described in the Kafka documentation.
Configs
configOverrides replaces all configs set on the topic.
Values are strings, so numbers must be quoted.
Configs that are set on the topic but missing in configOverrides are reset to the broker defaults.
Kafka applies all config changes of a topic together, so a change it rejects blocks the other changes until it is corrected.
The throttles of partition reassignments, leader.replication.throttled.replicas and follower.replication.throttled.replicas, are left untouched and cannot be set in configOverrides.
Configs changed outside of the KafkaTopic are reverted within five minutes, for example changes made with kafka-configs.sh.
Existing topics
A KafkaTopic for a topic that already exists in Kafka takes over the topic.
Taking over a topic is destructive and cannot be undone.
The agent treats a topic that existed before the KafkaTopic exactly like a topic it created itself.
The configs of the topic are replaced by configOverrides, which can delete data, for example with a shorter retention.ms or another cleanup.policy.
With deletionPolicy: Delete, deleting the KafkaTopic deletes the topic and all its data, even if both existed long before the KafkaTopic.
Neither the agent nor the operator keeps a backup of the topic data.
See Taking over a topic as it is for a takeover without changes.
|
A KafkaTopic refers to its topic by name. If the topic is deleted and created again outside of the KafkaTopic, the KafkaTopic takes over the new topic as well.
If several KafkaTopics declare the same topic in the same cluster, the oldest KafkaTopic manages the topic, and the others report a Conflict.
When the oldest KafkaTopic is deleted, the next oldest one takes over the topic within five minutes and applies its own spec to it.
What it takes over depends on the deletionPolicy of the deleted KafkaTopic, see Deletion:
-
Retain: the topic and its data stay in Kafka and are adopted. -
Delete: the topic is deleted first, and the next KafkaTopic creates a new, empty topic.
To hand a topic over to another KafkaTopic without losing data, set deletionPolicy: Retain on the old KafkaTopic before deleting it.
Taking over a topic as it is
To take over a topic without changing it, let the agent show its current state first:
-
Create the KafkaTopic for the existing topic with
topicOperation.reconciliationPaused: true, see Pausing reconciliation. The agent only observes the topic and fills the status. -
Copy
status.configintoconfigOverrides, andstatus.partitionsandstatus.replicationFactorinto the spec, see Status. Leave out the replication throttles of a running reassignment, see Configs. -
Set
reconciliationPausedtofalse. The topic matches the spec, so the agent changes nothing.
Status
The status shows the topic in Kafka:
status:
conditions: ...
topicId: UXyU9i5ARn6W00ON2taeWA (1)
topicName: shop.orders (2)
partitions: 6
replicationFactor: 3 (3)
config: (4)
cleanup.policy: delete
retention.ms: "604800000"| 1 | The topic ID as shown by kafka-topics.sh. |
| 2 | The name of the topic in Kafka. |
| 3 | The replication factor, excluding replicas that a running reassignment removes. It is not set if the partitions have different replication factors. |
| 4 | The configs set on the topic, without broker and default values. |
While the agent cannot reach Kafka, the status keeps the last observed state.
The Ready condition shows whether the topic exists in Kafka, and the Synced condition whether the topic matches the KafkaTopic.
The reason and message of a condition explain its status:
| Condition | Status | Reasons |
|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Pausing reconciliation
The reconciliation of a single KafkaTopic can be paused:
spec:
topicOperation:
reconciliationPaused: trueWhile paused, the agent does not change the topic in Kafka, but still updates the status.
Pausing the KafkaCluster with its cluster operations does not pause its KafkaTopics. A stopped KafkaCluster has no agent, so its KafkaTopics are not reconciled, see Deletion.
Deletion
With the default deletionPolicy: Retain, deleting a KafkaTopic leaves the topic in Kafka.
With deletionPolicy: Delete, the agent deletes the topic in Kafka when the KafkaTopic is deleted, also if a conflicting KafkaTopic declares the same topic.
A conflicting KafkaTopic never deletes the topic.
To hand a topic over to another KafkaTopic instead, switch the old one to deletionPolicy: Retain before deleting it.
The KafkaTopic gets a finalizer once the agent manages the topic, so the deletion waits until the topic is deleted.
If the reconciliation of the KafkaTopic is paused, the deletion waits until it is resumed.
If the brokers cannot be reached, the KafkaTopic reports ConnectionFailed and the agent retries the deletion.
Switching the KafkaTopic to deletionPolicy: Retain releases the deletion.
Switching an existing KafkaTopic to Delete applies to the topic as it is, including data that existed before the KafkaTopic.
With deletionPolicy: Delete every deletion of the KafkaTopic deletes the topic and all its data.
This includes adopted topics that existed before the KafkaTopic and deletions by tools that prune resources such as Argo CD or Flux.
|
When the KafkaCluster or its namespace is deleted, the operator removes the finalizers of its KafkaTopics, so the deletion does not block. The topics are not deleted in Kafka in this case, so a KafkaCluster created again with the same PersistentVolumeClaims still has them.
If the KafkaCluster exists but has no agent, for example because management access is disabled or the KafkaCluster is stopped, the deletion waits for the agent.
In this case, the finalizer kafka.stackable.tech/kafkatopic can be removed manually, which leaves the topic in Kafka.
A KafkaTopic that no agent has managed yet has no finalizer and is deleted at once.