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

Create a topic as a KafkaTopic CR:

apiVersion: platform.confluent.io/v1beta1
kind: KafkaTopic
metadata:
  name:              ----- [1]
  namespace:
spec:
  replicas:
  partitionCount:
  kafkaClusterRef:
    name:            ----- [2]
  kafkaRestClassRef:
    name:            ----- [3]
  kafkaRest:
    endpoint:        ----- [4]
    authentication:
       type:         ----- [5]
  • [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 ignored.

  • [2] Name of the Kafka cluster.

  • [3] Name of the KafkaRestClass CR.

  • [2] [3] If you do not provide kafkaRestClassRef and kafkaClusterRef in the topic CR, CFK tries to discover the Kafka cluster in the same namespace through the kafka CR type.

  • [4] Kafka endpoint. See Manage Kafka Admin REST APIs for more information.

  • [5] If authentication is required for Admin REST APIs, specify the authentication type. basic and bearer are supported.

    During discovery, if authenticationType is found in the Kafka cluster, the spec.kafkaRest.authentication must match in the topics CR configuration. Otherwise, the topic is not created.

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.

apiVersion: platform.confluent.io/v1beta1
kind: KafkaTopic
metadata:
  name: topic-a           ----- [1]
  namespace: confluent
spec:
  replicas: 1
  partitionCount: 12
  kafkaClusterRef:
    name: kafka
  kafkaRest:
    endpoint: https://kafka.confluent.svc.cluster.local:8090
    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.

You cannot update topics with the _ character in the topic name.

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>