<a id="co-sbc"></a>

# Scale Confluent Platform Clusters and Balance Data using Confluent for Kubernetes

For meeting changing workload demands and ensuring high availability and
performance, you can use Confluent for Kubernetes (CFK) to easily scale your Apache Kafka® clusters
and other Confluent Platform components up or down. CFK leverages the Self-Balancing Clusters
(Self-Balancing) features to automate data rebalancing and optimize resource utilization.
This guide provides the requirements and workflows for scaling cluster size and
data replication in CFK.

## Scale Kafka cluster

At a high level, adding brokers to a cluster involves a few key steps:

* Define the configuration for each of the new brokers.
* Provision storage, networking, and compute resources to the brokers.
* Start the brokers with the defined configurations and provisioned resources.
* Reassign partitions across the cluster so that the new brokers share the load
  and the cluster’s overall performance improves.

To automate the preceding process, CFK leverages Self-Balancing, which is
enabled by default with CFK.

If you need to manually disable or re-enable Self-Balancing, see [Disable or re-configure Self-Balancing](#co-enble-sbc) for
the steps.

<a id="co-cluster-scaling-limitations-details"></a>

### Requirements and considerations

Review the following requirements, considerations, and limitations for scaling
Kafka clusters and managing data replication in CFK.

**Scaling down or reducing brokers**
: When scaling down a Kafka cluster using CFK, do not perform
  any other operations on the cluster until the shrink process has completed
  successfully.
  <br/>
  #### WARNING
  Do not modify CFK cluster-shrink annotations such as `platform.confluent.io/enable-shrink` and `platform.confluent.io/cluster-scale-metadata`
  to work around a slow or stuck shrink.
  <br/>
  Editing these annotations can leave the cluster in an inconsistent state or hide an incomplete shrink.
  If a shrink is stuck, for troubleshooting guidance, see [If a shrink appears stuck](#co-scale-cluster-shrink-stuck). Contact [Confluent support](https://support.confluent.io/) for a guided recovery.
  <br/>
  * Scaling down removes brokers starting from the highest numbered broker.
  * You cannot remove brokers from the middle of a cluster. For example, in a
    6-broker setup, you cannot remove broker 2.
  <br/>
  For detailed broker removal procedures, see [Self-Balancing Clusters Configuration Options](https://docs.confluent.io/platform/current/clusters/sbc/configuration-options.html).

**When scaling up (adding brokers)**
: Expansion always adds brokers starting from the next highest number after the
  current highest numbered broker. For example, scaling from 6 to 7 brokers will
  create broker 6 (if brokers are numbered 0-5).

<a id="co-replication-throttling"></a>

#### Data replication throttling configuration

When Kafka clusters require data recovery through replication, you can fine-tune
throttling based on your cluster configuration:

| Property                                           | Description                                                                                                                                                                                                                                                      | Default Value   | Recommended Value                                                                                                                                                                                                                                                                         |
|----------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `follower.replication.throttled.rate`              | Upper bound (bytes/sec) on replication traffic for the followers<br/>enumerated in the property, `follower.replication.throttled.replicas`<br/>(for each topic).                                                                                                 | `unlimited`     | This property can only be set dynamically. Keep the limit above 1MB/sec<br/>for accurate behavior.                                                                                                                                                                                        |
| `leader.replication.throttled.rate`                | Upper bound (bytes/sec) on replication traffic for leaders enumerated in<br/>the property `leader.replication.throttled.replicas` (for each topic).                                                                                                              | `unlimited`     | This property can only be set dynamically. Keep the limit above 1MB/sec<br/>for accurate behavior.                                                                                                                                                                                        |
| `confluent.broker.limit.producer.bytes.per.second` | Broker-wide produce bandwidth limit (bytes/sec). The broker will<br/>throttle produce requests to ensure that the cumulative bandwidth does<br/>not exceed this limit.                                                                                           | `unlimited`     | Set to 20MB/sec for producer bandwidth.                                                                                                                                                                                                                                                   |
| `confluent.broker.limit.consumer.bytes.per.second` | Broker-wide consumer bandwidth limit (bytes/sec). The broker will<br/>throttle fetch requests to ensure that the cumulative bandwidth does not<br/>exceed this limit.                                                                                            | `unlimited`     | Set to 60MB/sec for consumer bandwidth.                                                                                                                                                                                                                                                   |
| `confluent.quota.tenant.broker.max.producer.rate`  | Maximum producer quota (bytes/sec) per tenant per broker.<br/><br/>This maximum is universal and applies to both followers and leaders.<br/>However, only brokers hosting at least one leader of the tenant’s topic<br/>partitions ever reach the maximum quota. | `12.5 MiB`      | Default cap on tenant quota that can be assigned to a single broker for<br/>produce and consume quotas: 12.5MB/sec each. With 100MB/sec cluster-wide<br/>tenant quota, a tenant needs to send load to at least 8 partitions (on 8<br/>brokers) to get the full produce and consume quota. |
| `confluent.quota.tenant.broker.max.consumer.rate`  | Maximum consumer quota (bytes/sec) per tenant per broker.<br/><br/>This maximum is universal and applies to both followers and leaders.<br/>However, only brokers hosting at least one leader of the tenant’s topic<br/>partitions ever reach the maximum quota. | `12.5 MiB`      | Default cap on tenant quota that can be assigned to a single broker for<br/>produce and consume quotas: 12.5MB/sec each. With 100MB/sec cluster-wide<br/>tenant quota, a tenant needs to send load to at least 8 partitions (on 8<br/>brokers) to get the full produce and consume quota. |

<a id="co-sbc-throttling"></a>

#### Self-Balancing clusters throttling configuration

#### IMPORTANT
Do not increase the disk size only for a particular broker. Disks should be the same size across brokers as a precondition for Self-Balancing. A discrepancy of disk size among the brokers causes the DiskUsageDistributionGoal violation.

For Self-Balancing Clusters, configure the following throttling properties. All
the property names are prefixed with `confluent.balancer`, such as
`confluent.balancer.producer.in.max.bytes.per.second`.

| Property                              | Description                                                                                                                                                                                                             | Default Value         | Recommended Value                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
|---------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `producer.in.max.bytes.per.second`    | Upper limit for producer incoming bytes/sec/broker. Self-Balancing will attempt to<br/>keep incoming data throughput below this limit.                                                                                  | `9223372036854775807` | If the property is set higher than<br/>`confluent.broker.limit.producer.bytes.per.second`, throttling will<br/>be set earlier, and it may not reach this limit.<br/><br/>If it is set lower than the<br/>`confluent.broker.limit.producer.bytes.per.second` limit, it will make<br/>Self-Balancing attempt to keep the producer incoming traffic under the proposed<br/>value, and you will have a certain headroom reserved, which can be used<br/>for cluster maintenance tasks without affecting their primary produce<br/>traffic. However, setting a value that is just 10-15% above the produce<br/>traffic seen by the cluster will lead to replica movements.<br/><br/>Setting it at least 20% under the<br/>`confluent.broker.limit.producer.bytes.per.second` limit is advised to<br/>avoid unnecessary movements while still providing a safe upper limit<br/>without hitting a throttle. |
| `consumer.out.max.bytes.per.second`   | Upper limit for consumer outgoing bytes/sec/leader broker.<br/><br/>Self-Balancing will attempt to keep outgoing data throughput below this limit.<br/><br/>Note that Fetch From Follower traffic is not accounted for. | `9223372036854775807` | It has the same semantics and behavior as<br/>`confluent.balancer.producer.in.max.bytes.per.second`, but it is<br/>associated with `confluent.broker.limit.consumer.bytes.per.second`.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
| `replication.in.max.bytes.per.second` | Upper limit for replication incoming bytes/sec/broker.<br/><br/>Self-Balancing will attempt to keep incoming replication traffic below this limit.                                                                      | `9223372036854775807` | The replication inbound traffic has the same caveats as<br/>`confluent.balancer.producer.in.max.bytes.per.second`. However, you<br/>must be conscious of the replication factor of your cluster and<br/>ensure the value is not less than `(replication factor - 1)<br/>* confluent.balancer.producer.in.max.bytes.per.second` because ISR<br/>traffic is accounted for here.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |

### Scale up Kafka cluster

To scale up a Kafka cluster:

1. Increase the number of Kafka replicas using one of the following options:
   * Use the `kubectl scale` command:
     ```bash
     kubectl scale kafka <Kafka-CR-name> --replicas=N
     ```
   * Increase the number of Kafka replicas in the Kafka custom resource (CR) and
     apply the new setting with the `kubectl apply` command:
     ```yaml
     spec:
       replicas:
     ```
2. Ensure that proper DNS records are configured for the new brokers, and ensure
   that the CFK can resolve the new broker hostname, using a command such as
   `nslookup`.

   If you are using `hosts` file instead of a DNS service, update `hosts`
   file with the new brokers information. For example:
   1. Get the new broker IP addresses:
      ```bash
      kubectl get services -n <namespace>
      ```
   2. Refer to the existing broker host names with the broker prefix, and derive
      the hostnames of the new brokers.
   3. Add the new broker hosts to the `/etc/hosts` file, and inject the
      updated file to the CFK pod as described in
      [Adding entries to Pod /etc/hosts](https://kubernetes.io/docs/concepts/services-networking/add-entries-to-pod-etc-hosts-with-host-aliases/#adding-additional-entries-with-hostaliases).

### Scale down Kafka cluster

With Confluent Platform 7.x and later, you can have CFK scale down Kafka clusters.

CFK leverages the Self-Balancing feature to automate the shrinking process. Self-Balancing is
enabled by default with CFK.

#### NOTE
Due to a [known limitation in Kubernetes scheduler](https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints/#known-limitations),
in multi-rack / availability zone (AZ) clusters, Kafka cluster shrinking is not
supported.

To have CFK automatically scale down your cluster, the following requirements
must be satisfied:

* Set up the Admin REST Class as described in [Manage Confluent Admin REST Class for Confluent Platform Using Confluent for Kubernetes](co-manage-rest-api.md#co-manage-rest-api). CFK
  uses the `KafkaRestClass` resource in the namespace where the Kafka cluster is
  running.

  For example `KafkaRestClass` configuration, see the [production-secure-deploy
  example](https://github.com/confluentinc/confluent-kubernetes-examples/blob/master/security/production-secure-deploy/kafkaRestClass.yaml).
* Configure `spec.dependencies.kafkaRest` in the Kafka custom resource (CR) to
  provide the credentials that CFK uses to call the Admin REST API as a client.
  This is required for CFK to perform cluster management operations including
  scale down and dynamic broker configuration updates.

  #### IMPORTANT
  The Kafka CR has two separate `kafkaRest` configurations that serve
  different purposes:
  * `spec.services.kafkaRest`: Configures the embedded REST API server
  * `spec.dependencies.kafkaRest`: Provides the credentials CFK uses to
    call that API as a client

  These are independent configurations. The server-side configuration does not
  automatically provide client credentials. Both must be configured for scale
  down operations.

  When using OAuth authentication, ensure that the `scope` parameter is
  configured in `spec.dependencies.kafkaRest.authentication` in addition to
  `spec.services.kafkaRest`. A missing or incorrect scope in the client
  configuration will cause scale down operations to fail, even if the server is
  configured correctly.
* If the Admin REST Class is set up with the basic authentication for the REST
  client, the first user listed in `basic.txt` will be used to scale down the
  cluster. See [Basic authentication](co-authenticate-cp.md#co-authenticate-cp-basic) for details on `basic.txt`.

  This first user must have a role that is listed under
  `spec.services.kafkaRest.authentication.basic.roles` in the Kafka custom
  resource (CR).
* If the Kafka brokers use the `DirectoryPathInContainer` property to specify
  the credentials to authenticate to Confluent Admin REST Class, you need to set
  up Vault and add the required Vault annotations to the CFK Helm values before
  you deploy CFK.

  If updating an existing CFK pod, you need to roll the CFK  pod after
  updating the CFK Helm values. See [Provide secrets for Confluent Platform operations without CRs](co-credentials.md#co-vault-category-3) for  details.
* When performing the scale down/shrink operation with CFK, keep the webhooks
  disabled. To disable the webhooks, there are two options:
  * [Disable all the webhooks](co-cfk-settings.md#co-disable-webhooks).
  * If you don’t want to disable all the webhooks, and want to scale down the
    cluster, consider disabling the [webhook for pod deletion](co-cfk-settings.md#co-disable-webhooks).

  Webhooks would cause an issue with the cluster shrink workflow. Keeping
  webhooks enabled and performing the cluster shrink at the same time causes a
  mismatch between the expected replica and the actual replica, which results in
  a stuck reconcile loop.

To automatically scale down a Kafka cluster:

1. Make sure the Kafka cluster is stable.
2. Decrease the number of brokers in the Kafka CR and apply the change using the
   `kubectl apply` command:
   ```yaml
   spec:
     replicas:
   ```

   `replicas:` should not be set to less than `3`. CFK sets a default
   replication factor of 3 for all Kafka topics.

   CFK triggers the workflow to shrink the Kafka cluster according to the value
   of `replicas` updated in the Kafka custom resource (CR).

<a id="co-scale-cluster-shrink-stuck"></a>

#### If a shrink appears stuck

If an Kafka cluster shrink triggered by CFK seems stuck, review the
following guidance:

* Verify that the prerequisites described in
  [Requirements and considerations](#co-cluster-scaling-limitations-details) are met, particularly
  that the Kafka cluster is healthy and the required webhooks are
  disabled.
* Check the `status.conditions` of the Kafka CR for a condition of
  type `platform.confluent.io/cluster-shrink`, and review the CFK
  operator logs to confirm whether the shrink workflow is still being
  reconciled.
* Collect a support bundle along with a description of the Kafka CR,
  including its `status` and annotations, and recent CFK operator
  logs, and open a case with [Confluent support](https://support.confluent.io/) for a guided recovery.
* Do not attempt to fix a stuck shrink by manually setting
  `platform.confluent.io/enable-shrink` to `false` or `true`, or
  by overwriting the `platform.confluent.io/cluster-scale-metadata`
  annotation. Editing these annotations can leave the cluster in an
  inconsistent state or hide an incomplete shrink.

<a id="co-disable-ak-cluster-shrinking"></a>

### Disable scaling down of Kafka cluster

The Kafka cluster shrinking feature is enabled by default.

To disable the cluster shrinking feature, apply the annotation,
`platform.confluent.io/enable-shrink=false`, to the Kafka cluster:

```bash
kubectl annotate Kafka <Kafka CR name> -n <namespace> platform.confluent.io/enable-shrink=false
```

<a id="co-enble-sbc"></a>

### Disable or re-configure Self-Balancing

The Self-Balancing feature is enabled by default in Confluent for Kubernetes.

To balance the load across the cluster whenever an imbalance is detected, set
[confluent.balancer.heal.uneven.load.trigger](/platform/current/kafka/sbc/configuration_options.html#sbc-config-heal-uneven-load)
to `ANY_UNEVEN_LOAD`. The default is `EMPTY_BROKER`.

```yaml
kind: Kafka
spec:
  configOverrides:
    server:
      - confluent.balancer.heal.uneven.load.trigger=ANY_UNEVEN_LOAD
```

```yaml
kind: KRaftController
spec:
  configOverrides:
    server:
      - confluent.balancer.heal.uneven.load.trigger=ANY_UNEVEN_LOAD
```

For a complete list of available settings you can use to control Self-Balancing, see
[Configuration Options and Commands for Self-Balancing
Clusters](/platform/current/kafka/sbc/configuration_options.html#sbc-config-options).

## Scale other Confluent Platform components

Other Confluent Platform component clusters, namely ksqlDB, Connect, and REST Proxy
clusters, can be scaled up or down depending on the load. Schema Registry does not have use
cases for scaling.

#### IMPORTANT
Scaling a ZooKeeper ensemble up or down after initial deployment is not
supported in CFK. Do not use `kubectl scale` or change the
`spec.replicas` value on a ZooKeeper CR to resize an
existing ensemble. If you do this, it can destabilize the ensemble. Set the
number of ZooKeeper replicas only at initial deployment.

When scaling REST Proxy, it is important that you use either a sticky load balancer
or session affinity. For details, see [Session affinity for REST Proxy
consumers](co-configure-rest-proxy.md#co-rest-proxy-session-affinity).

Scale up or down other Confluent Platform components using one of the following options:

* Use the `kubectl scale` command:
  ```bash
  kubectl scale <CP-component-CR-kind> <component-CR-name> --replicas=N
  ```
* Update the number of replicas in the component CR and
  apply the new setting with the `kubectl apply` command:
  ```yaml
  kind: <CP component>
  spec:
    replicas: <new number of replicas>
  ```