Manage Kafka Topics¶

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.

KafkaTopic communicates with a Kafka cluster to create, edit, and delete topics through Admin REST APIs. You need to set up Admin REST APIs as described in Kafka Admin REST API before you create KafkaTopic CRs.

Create Kafka topic¶

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

kind: KafkaTopic
metadata:
  name:              ----- [1]
  namespace:
spec:
  replicas:
  partitionCount:
  kafkaClusterRef:     --- [2]
  kafkaRestClassRef:   --- [3]
  kafkaRest:
    endpoint:          --- [4]
    kafkaClusterID:    --- [5]
    authentication:
       type:           --- [6]

[1] The topic name. The _ character in the topic name is not allowed.

If both metadata.name and spec.name are specified, spec.name is used.
Use kafkaClusterRef ([2]), kafkaRestClassRef ([3]), or kafkaRest.endpoint ([4]) to explicitly specify the Admin REST APIs.

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

If none of the above is set, it performs an auto discovery of the Kafka in the same namespace.
[2] Name of the Kafka cluster.
[3] Name of the KafkaRestClass CR.
[4] Admin REST APIs endpoint. See Manage Confluent Admin REST API for more information.
[5] ID of the Kafka cluster. Required when creating a topic in Confluent Cloud.
[6] If authentication is required for Admin REST APIs, specify the authentication type. basic, bearer, mtls are supported.

If you specified the Admin REST APIs using kafkaRestClassRef, you do not have to set the authentication in kafkaRest. Otherwise specify the authentication in kafkaRest.

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.

kind: KafkaTopic
metadata:
  name: topic-a
  namespace: confluent
spec:
  replicas: 1
  partitionCount: 12
  kafkaClusterRef:
    name: kafka
  kafkaRest:
    authentication:
      type: bearer
      bearer:
        secretRef: rest-credential

For the list of available topics configuration parameters, see Kafka Topics Configurations.

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:

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

Warning

spec.replicas and spec.partitionCount cannot be updated.

For the list of available topics configuration parameters, see Kafka Topics Configurations.

Delete Kafka topic¶

To delete a topic, use the kubectl delete command.

For example:

kubectl delete -f <topic-cr-file>