<a id="co-manage-topics"></a>

# Manage Kafka Topics for Confluent Platform Using Confluent for Kubernetes

Confluent for Kubernetes (CFK) allows you to declaratively create and manage Kafka topics as
KafkaTopic custom resources (CRs) in Kubernetes. Each KafkaTopic CR is mapped to
a topic and kept in sync with the corresponding Kafka topic. This allows you to
have a separate workflow where you create topics as part of Confluent
deployment, and your client applications only need to produce and consume from
the topics.

#### NOTE
KafkaTopic CRs stay in sync with the Kafka topics if created and modified with
KafkaTopic CRs. If you modify topic configuration outside of the KafkaTopic
CR, such as using Control Center or Confluent CLI, the change is not reconciled, and
the KafkaTopic CR and the topic configuration get out of sync.

## Requirement

KafkaTopic communicates with a Kafka cluster through the Confluent REST Class to
create, edit, and delete topics. You need to set up Kafka Admin REST Class as
described in [Kafka Admin REST Class](co-manage-rest-api.md#co-manage-rest-api) before you
create KafkaTopic CRs.

#### IMPORTANT
On an RBAC-enabled cluster, you must set `kafkaRestClassRef` on the
KafkaTopic CR. Without it, CFK does not fall back to a `default`
KafkaRestClass CR. For details about the order of precedence, see
[Create Kafka topic](#co-manage-topics-create-ak-topic).

<a id="co-manage-topics-create-ak-topic"></a>

## Create Kafka topic

You can create a topic using a KafkaTopic CR in an on-prem or Confluent Cloud Kafka
cluster:

```yaml
kind: KafkaTopic
metadata:
  name:                --- [1]
  namespace:           --- [2]
spec:
  replicas:
  partitionCount:
  kafkaClusterRef:     --- [3]
  kafkaRestClassRef:   --- [4]
  kafkaRest:
    endpoint:          --- [5]
    kafkaClusterID:    --- [6]
    authentication:
       type:           --- [7]
       basic:          --- [8]
       bearer:         --- [9]
       oauth:          --- [10]
  configs:             --- [11]
```

* [1] The topic name. If both `metadata.name` and `spec.name` are specified,
  `spec.name` is used.
* [2] The namespace for the topic.
* Use `kafkaClusterRef` ([3]), `kafkaRestClassRef` ([4]), or
  `kafkaRest.endpoint` ([5]) to explicitly specify the Confluent REST Class.

  The order of precedence is [4], [5], and [3].

  If neither [4] nor [5] is set, CFK auto-discovers the REST endpoint
  directly from the Kafka cluster referenced by `kafkaClusterRef` ([3]).
  Unlike ConfluentRolebinding, this auto-discovery does not fall back to
  a `default` KafkaRestClass CR, even if one exists in the same namespace.

  Auto-discovery only succeeds if the referenced Kafka cluster’s Admin REST APIs
  does not require authentication. If the cluster is RBAC-enabled and
  neither `kafkaRestClassRef` nor `kafkaRest.authentication` is set,
  topic reconciliation fails with an error like:
  ```text
  kafkatopic apply failed: authentication is enabled for kafka rest api for a cluster [<cluster-name>]
  ```

  To avoid this on an RBAC-enabled cluster, set `kafkaRestClassRef` ([4])
  to reference a KafkaRestClass CR, for example, one named `default`.
  Alternatively, specify authentication inline under `kafkaRest`
  ([7]-[10]).
* [3] Name of the Kafka cluster.
* [4] Name of the KafkaRestClass CR.
* [5] Confluent REST Class endpoint. See [Manage Confluent Admin REST Class for Confluent Platform Using Confluent for Kubernetes](co-manage-rest-api.md#co-manage-rest-api) for more
  information.
* [6] ID of the Kafka cluster. Required when creating a topic in Confluent Cloud.
* [7] If authentication is required for the Confluent Admin REST Class, specify
  the authentication type. `basic`, `bearer`, `mtls`, and `oauth` are
  supported.

  If you specified the Confluent Admin REST Class using `kafkaRestClassRef`,
  you do not have to set the authentication in `kafkaRest`. Otherwise specify
  the authentication in `kafkaRest`.
* [8] For information about the basic settings, see
  [Basic authentication](co-authenticate-cp.md#co-authenticate-cp-basic).
* [9] For information about the bearer settings, see
  [Bearer authentication](co-authenticate-kafka.md#co-authenticate-mds-bearer),
* [10] For information about the OAuth settings, see
  [OAuth/OIDC authentication](co-authenticate-cp.md#co-authenticate-cp-oauth).
* [11] Specify additional topic configuration settings in key and value pairs,
  for example, `cleanup.policy: "compact"`.

  For the list of available topics configuration parameters, see [Kafka Topics
  Configurations](https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html).

#### IMPORTANT
After you create a topic, you should NOT change the Kafka cluster used by that
topic CR. That can lead to possible data loss.

The example CR below creates a Kafka topic, named `topic-a`, in the
`confluent` namespace with `1` replica and `12` partitions. The topic is
created in the internal Kafka cluster, `kafka`, with the `bearer`
authentication.

```yaml
kind: KafkaTopic
metadata:
  name: topic-a
  namespace: confluent
spec:
  replicas: 1
  partitionCount: 12
  configs:
    cleanup.policy: "compact"
  kafkaClusterRef:
    name: kafka
  kafkaRest:
    authentication:
      type: bearer
      bearer:
        secretRef: rest-credential
```

<a id="co-manage-topics-update-kafka-topic"></a>

## Update Kafka topic

To update a topic, change the topic CR configuration, and apply the changes
using the `kubectl apply` command.

The following example adds a cleanup policy to the above `topic-a` topic:

```yaml
apiVersion: platform.confluent.io/v1beta1
kind: KafkaTopic
metadata:
  name: topic-a
  namespace: confluent
spec:
  replicas: 1
  partitionCount: 12
  configs:
    cleanup.policy: "compact"
```

#### WARNING
You cannot update `spec.replicas` after the topic is created.

For Confluent for Kubernetes 3.2.0 and later, you can increase `spec.partitionCount`
for an existing topic by editing the KafkaTopic custom resource. Decreasing `spec.partitionCount` is not supported.

To increase the partition count for an existing topic, update only `spec.partitionCount` to a larger value and re-apply the CR:

```yaml
apiVersion: platform.confluent.io/v1beta1
kind: KafkaTopic
metadata:
  name: topic-a
  namespace: confluent
spec:
  replicas: 1
  partitionCount: 24
  configs:
    cleanup.policy: "compact"
```

CFK reconciles the CR and increases the Kafka topic’s partition count to match the new value.

#### IMPORTANT
When upgrading existing (brownfield) deployments to CFK 3.2.0 or later,
verify partition counts across all KafkaTopic CRs before upgrading. If
`spec.partitionCount` differs from the actual partition count in the
cluster, CFK reconciles the partition count on the first operator run
after upgrade, which may result in an unintended partition increase.

For the list of available topics configuration parameters, see [Kafka Topics
Configurations](https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html).

## Delete Kafka topic

To delete a topic, use the `kubectl delete` command.

For example:

```bash
kubectl delete -f <topic-cr-file>
```