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

[id='proc-converting-non-managed-topics-{context}']
= (Preview) Enabling topic management for existing Kafka topics

[role="_abstract"]
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 xref:ref-operator-unidirectional-topic-operator-feature-gate-{context}[enable the `UnidirectionalTopicOperator` feature gate] to be able to use it.

.Prerequisites

* xref:deploying-cluster-operator-str[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 Kubernetes resource name.
+
.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
----
+
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:
+
[source,shell]
----
kubectl 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:
+
[source,shell,subs="+quotes"]
----
oc get kafkatopics my-topic-1 -o yaml
----
+
.Example topic with a `Ready` status
[source,shell,subs="+attributes"]
----
apiVersion: {KafkaTopicApiVersion}
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).