Confluent Developer Marketplace Get started
  1. Home
  2. Platform
  3. CFK
  4. Deploy and Operate Confluent Platform with Confluent for Kubernetes
  5. Manage Confluent Platform with Confluent for Kubernetes

Manage Kafka Topics for Confluent Platform Using Confluent for Kubernetes

Confluent for Kubernetes (CFK) allows you to declaratively create and manage Apache 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.

For Confluent for Kubernetes 3.4.0 and later, you can change this behavior with the platform.confluent.io/enable-config-reset and platform.confluent.io/enable-drift-correction annotations. See Annotate Confluent custom resources for details.

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 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.

Create Kafka topic

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

kind: KafkaTopic
metadata:
  name:                --- [1]
  namespace:           --- [2]
spec:
  replicas:
  partitionCount:
  kafkaClusterRef:     --- [3]
    name:              --- [3a]
    namespace:         --- [3b]
    kind:              --- [3c]
  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:

    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] A reference to the Kafka cluster.

    • [3a] Required under kafkaClusterRef ([3]). The name of the Kafka CR. On a Confluent Private Cloud multi-tenant Physical Kafka Cluster (PKC), you can set this to the name of a LogicalKafkaCluster CR instead, scoping the topic to that single Logical Kafka Cluster (LKC).

    • [3b] Optional. The namespace of the referenced CR. If omitted, defaults to the namespace of this KafkaTopic CR.

    • [3c] Optional. Set to Kafka or LogicalKafkaCluster to disambiguate when a Kafka CR and a LogicalKafkaCluster CR share the same name ([3a]) in the namespace. If kind is unset and both exist, CFK fails to reconcile the CR and reports an ambiguous reference error in the CR status.

      For more information, see Onboard a Logical Kafka Cluster in Confluent Private Cloud.

  • [4] Name of the KafkaRestClass CR.

  • [5] Confluent REST Class endpoint. See Manage Confluent Admin REST Class for Confluent Platform Using Confluent for Kubernetes 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.

  • [9] For information about the bearer settings, see Bearer authentication.

  • [10] For information about the OAuth settings, see OAuth/OIDC authentication.

  • [11] Specify additional topic configuration settings in key and value pairs, for example, cleanup.policy: "compact".

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

Important

After you create a topic, you must not change the Kafka cluster used by that topic CR. That can lead to possible data loss.

The following example CR 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
  namespace: confluent
spec:
  replicas: 1
  partitionCount: 12
  configs:
    cleanup.policy: "compact"
  kafkaClusterRef:
    name: kafka
  kafkaRest:
    authentication:
      type: bearer
      bearer:
        secretRef: rest-credential

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 preceding topic-a topic:

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

For Confluent for Kubernetes 3.4.0 and later, you can update spec.replicas to change an existing topic’s replication factor. For more information, see Change the replication factor of a topic.

Note

For Confluent for Kubernetes 3.3.x and earlier, 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:

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 topic configuration parameters, see Kafka Topics Configurations.

Change the replication factor of a topic

Starting in CFK 3.4.0 with Confluent Platform 8.4 or later, you can change the replication factor (RF) of an existing topic by editing spec.replicas on its KafkaTopic CR. You can increase or decrease the RF. CFK hands the change to Self-Balancing Clusters, which chooses the brokers and moves the data using partition reassignment. CFK does not move data or validate spec.replicas against brokers, racks, or min.insync.replicas. Self-Balancing Clusters validates the request and returns an error code if the request is not valid.

Requirements

  • On every KRaftController CR, set spec.listeners.replication to match the replication listener of the Kafka brokers. Without it, Self-Balancing Clusters cannot reach the broker listeners.

    If the brokers and the controllers use different authentication mechanisms on the replication listener, also set sasl.mechanism.inter.broker.protocol to the mechanism the brokers use.

  • On every KRaftController CR, set the following override to enable RF changes in Self-Balancing Clusters.

    kind: KRaftController
spec:
  configOverrides:
    server:
      - confluent.balancer.replication.factor.change.enabled=true

  • Provide a KafkaRestClass for the topic’s cluster. Set spec.kafkaRestClassRef on the topic, or create a KafkaRestClass named default in the topic’s namespace. Without a KafkaRestClass, CFK refuses the change, sets the topic to ERROR, and leaves the RF unchanged.

  • Grant the KafkaRestClass principal the Cluster:ALTER permission on the cluster. On an RBAC-enabled cluster, bind the ClusterAdmin role at cluster scope. Without it, Self-Balancing Clusters returns HTTP 403 with the error code 40301.

  • Make sure the brokers have enough disk space. An increase from RF n to m stores m - n extra full copies of the topic. During reassignment, data can also be present on additional brokers temporarily.

Change the replication factor

To change the RF, update spec.replicas and re-apply the CR. For example, the following CR increases the RF of orders to 4:

apiVersion: platform.confluent.io/v1beta1
kind: KafkaTopic
metadata:
  name: orders
spec:
  replicas: 4
  partitionCount: 6
  kafkaClusterRef:
    name: kafka

Keep the following in mind:

  • Do not set the target RF above the number of brokers that can hold a replica. Self-Balancing Clusters fails the change with the error code 10044. In multi-region and rack-aware clusters, fewer brokers can be eligible than exist.

  • Do not set the target RF below the min.insync.replicas value of the topic. Self-Balancing Clusters fails the change with the error code 10038. Lower min.insync.replicas in spec.configs first.

  • After a change fails with 10044 or 10038, the change stops. Edit spec.replicas, or add brokers and set the annotation platform.confluent.io/force-reconcile: "true" on the topic to retry the same value.

  • To change both partitionCount and replicas, make both edits in the same update. CFK adds the partitions first, and then moves every partition to the new RF.

  • Change topics in batches of a bounded size. Self-Balancing Clusters plans one topic at a time and moves no data until it plans the whole batch. Submit a batch, wait for it to finish, and then submit the next batch.

  • Do not delete a topic and recreate it with the same name while its change is pending. Self-Balancing Clusters tracks changes by topic name and can apply the old change to the new topic.

  • Do not remove spec.replicas to stop a change. Removing it does not cancel the change. It only stops CFK from managing the RF.

  • RF changes on internal topics, such as __consumer_offsets or topics that start with _confluent, are refused by default. Kafka creates internal topics automatically, so they have no KafkaTopic CR. To change the RF of an internal topic, take the following steps:

    1. Create a KafkaTopic CR for the internal topic. Set every topic configuration in the CR to match the live topic.

    2. Add the annotation platform.confluent.io/allow-internal-topic-replication-factor-change: "true" to the CR.

    3. Update spec.replicas to the new RF.

    The annotation is always required to change an internal topic. Without it, CFK refuses the change.

    Warning

    Do not delete the KafkaTopic CR of an internal topic. Deleting the CR deletes the internal topic, which can be dangerous.

    Do not enable the platform.confluent.io/enable-drift-correction annotation on the CR unless all of the topic configs are in the CR. Drift correction resets any config that the CR leaves out.

Monitor the change

CFK polls Self-Balancing Clusters every 10 minutes by default and records the result in the KafkaTopic status. The edit takes effect immediately, and the status can trail by up to one poll interval. To change the interval for all topics in a cluster, annotate the Kafka CR with platform.confluent.io/replication-factor-change-poll-interval-seconds. The minimum value is 30.

status:
  replicas: 4                       # Actual RF
  replicationFactorChange:
    state: Completed                # Pending, InProgress, Completed, or Failed
    action: Change                  # Change or Cancellation
    targetReplicationFactor: 4

A change is complete when state is Completed, action is Change, and targetReplicationFactor, spec.replicas, and status.replicas are equal. To check the actual RF of each partition, run the kafka-topics --describe command.

Self-Balancing Clusters reports status for a whole batch of topics. A topic can finish its RF change while another topic in the same batch is still pending. Self-Balancing Clusters then reports the batch as in progress, and CFK shows the finished topic as InProgress. As a result, the status of a KafkaTopic CR can lag behind the actual RF.

If a broker restarts or the active KRaft controller fails over during a change, Self-Balancing Clusters can interrupt the change. CFK keeps tracking the topic and resubmits the change on its next poll until the topic reaches spec.replicas. You do not need to take any action.

Go back or stop a change

To go back, set spec.replicas to the original value. CFK cancels the running change and then moves every partition to that value. This is the recommended way to reach a specific RF.

To halt a change for one topic in an emergency, set the annotation platform.confluent.io/replication-factor-change-hold: "true" on the KafkaTopic CR. A hold cancels the in-flight move and does not revert it, so the topic can end at the old RF, the target RF, or a mix across partitions. To release the hold, remove the annotation. CFK then converges the topic to spec.replicas.

Note

Any cancel stops the whole running Self-Balancing Clusters batch. Self-Balancing Clusters reverts the in-flight moves of the other topics in the batch and copies those partitions again.

Limitations

  • Topics in a LogicalKafkaCluster are not supported.

  • CFK recognizes the Confluent Platform version only for the stock confluentinc/cp-server image. Version gating does not apply to a mirrored or custom image. If you use one, make sure it is Confluent Platform 8.4 or later.

  • Do not increase the day-2 worker count above 1. RF changes are not supported with more than one day-2 worker.

  • Before you upgrade to CFK 3.4.0 with Confluent Platform 8.4.0 or later, check topics whose spec.replicas differs from status.replicas. CFK starts an RF change for each of these topics on the first reconcile after the upgrade. To keep the current RF, set spec.replicas to the actual RF before the upgrade. If you do not check, CFK still tries the RF change after the upgrade, but the RF does not change until you enable confluent.balancer.replication.factor.change.enabled=true on the KRaftController CR. CFK sets the platform.confluent.io/replication-factor-change-active annotation on each of these topics and keeps retrying the change. You can then either enable the override to complete the RF change, or revert the change. To revert, first set spec.replicas back to the actual RF, and then remove the platform.confluent.io/replication-factor-change-active annotation from the topics where CFK set it.

Delete Kafka topic

To delete a topic, use the kubectl delete command.

For example:

kubectl delete -f <topic-cr-file>