// Module included in the following assemblies:
//
// assembly-using-the-topic-operator.adoc

[id='proc-configuring-kafka-topic-{context}']
= Configuring Kafka topics

[role="_abstract"]
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 `kubectl apply` to create or modify topics, and `kubectl delete` to delete existing topics.

For example:

* `kubectl apply -f <topic_config_file>`
* `kubectl 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

. Configure the `KafkaTopic` resource.
+
.Example Kafka topic configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaTopicApiVersion}
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 `kubectl get kafkatopic my-topic-1 -o yaml`.

. Create the `KafkaTopic` resource in Kubernetes.
+
[source,shell,subs=+quotes]
kubectl apply -f _<topic_config_file>_

. Wait for the ready status of the topic to change to `True`:
+
[source,shell,subs="+quotes"]
----
kubectl get kafkatopics -o wide -w -n _<namespace>_
----
+
.Kafka topic status
[source,shell,subs="+quotes"]
----
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.
+
[source,shell,subs="+quotes"]
----
oc get kafkatopics my-topic-2 -o yaml
----
+
.Details on a topic with a `NotReady` status
[source,shell,subs="+quotes"]
----
# ...
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.
+
[source,shell,subs="+quotes"]
----
kubectl get kafkatopics my-topic-2 -o wide -w -n _<namespace>_
----
+
.Status update of the topic
[source,shell,subs="+quotes"]
----
NAME         CLUSTER     PARTITIONS  REPLICATION FACTOR READY
my-topic-2   my-cluster  10          3                  True
----
+
Fetching the details shows no messages
+
[source,shell,subs="+quotes"]
----
kubectl get kafkatopics my-topic-2 -o yaml
----
+
.Details on a topic with a `READY` status
[source,shell,subs="+quotes"]
----
# ...
status:
  conditions:
  - lastTransitionTime: '2022-06-13T10:15:03.761084Z'
    status: 'True'
    type: Ready
----
