<a id="sbc-config-options"></a>

# Self-Balancing Cluster Configuration and Command Reference for Confluent Platform

Self-Balancing configuration options and commands control how automatic
rebalancing behaves in your cluster.

See [Tutorial: Add and Remove Brokers with Self-Balancing in Confluent Platform](sbc-tutorial.md#sbc-tutorial) for an example of how to set up and run a quick test of Self-Balancing by removing a broker and
monitoring the rebalance. The example provides guidance on proper configuration of replication factors and demos
the Self-Balancing specific command, [kafka-remove-brokers](#sbc-command-remove-brokers).

<a id="sbc-configs-on-brokers"></a>

## Self-Balancing configuration

The following example configuration files are provided in the `etc/kafka` directory.

- `$CONFLUENT_HOME/etc/kafka/broker.properties`
- `$CONFLUENT_HOME/etc/kafka/controller.properties`
- `$CONFLUENT_HOME/etc/kafka/server.properties`

In most cases, if you specify a Self-Balancing configuration for brokers (`broker.properties`),
you must specify the same setting for the KRaft controllers (`controller.properties`) in the cluster.

If you are using Self-Balancing in combination with [Configure Multi-Region Clusters in Confluent Platform](../../multi-dc-deployments/multi-region.md#bmrr) , you must also specify the rack location for each broker with
[broker rack](/platform/current/installation/configuration/broker-configs.html#broker-rack) in each properties file.
To learn more, see [Replica placement and multi-region clusters](index.md#sbc-replicas-and-mrc).

<a id="sbc-config-enable"></a>

### confluent.balancer.enable

Enables self balancing, meaning load across the Kafka cluster is measured and data is rebalanced as needed, [depending on multiple goals and factors](index.md#sbc-rebalance-trigger-conditions).
If this property is specified as `false`, or not explicitly specified at all in the properties file, the value is inferred to be `false` or off.

In the example `server.properties`, `controller.properties` and `broker.properties` files for KRaft mode,  `confluent.balancer.enable=true` is commented out, which means Self-Balancing is off.
You must uncomment or add `confluent.balancer.enable=true` to the properties file for every broker and controller in the cluster to turn on Self-Balancing.
In addition, you must set `inter.broker.listener.name` for each broker and controller in the properties file. For more about configuring KRaft mode, see [Configuration options](../../kafka-metadata/config-kraft.md#kraft-config-options).

The Self-Balancing feature is [dynamic](#sbc-kafka-configs-command) meaning you can toggle Self-Balancing on or off while the cluster is running.

Do not disable Self-Balancing while an add or remove broker operation is in progress; wait until the add or remove completes. Disabling Self-Balancing during either operation
fences off the broker from future partition distribution.

* Type: boolean
* Default: false
* Importance: high

<a id="sbc-config-heal-uneven-load"></a>

### confluent.balancer.heal.uneven.load.trigger

Sets the conditions upon which to trigger a rebalance.
Valid values are:

* `EMPTY_BROKER`: Move data to rebalance the cluster only when an empty broker (one with no partitions on it) is added to the cluster.
* `ANY_UNEVEN_LOAD`: Balance the load across the cluster whenever an imbalance is detected. This includes empty brokers and any other uneven load
  condition caused by ongoing cluster activity. For example, load peaks on specific partitions
  could trigger a rebalance.

This is a [dynamic option](#sbc-kafka-configs-command), meaning
you can change the trigger condition for Self-Balancing while the cluster is running.

* Type: string
* Default: `EMPTY_BROKER`
* Importance: high

<a id="sbc-config-heal-broker-failure-threshold"></a>

### confluent.balancer.heal.broker.failure.threshold.ms

Sets the default time to declare a broker has permanently failed.
To adjust the threshold for the duration of time before a broker is declared “failed,”
edit `$CONFLUENT_HOME/etc/kafka/server.properties`.
Uncomment the line for `confluent.balancer.heal.broker.failure.threshold.ms`
and change its value to a positive number of your choosing. To turn off broker failure detection
uncomment the line that sets `confluent.balancer.heal.broker.failure.threshold.ms` to `-1`.

When a broker disappears from the cluster, Self-Balancing waits this length of time
before declaring the broker permanently failed and rebalancing its data onto
other brokers. This is independent of what value the
[confluent.balancer.heal.uneven.load.trigger](#sbc-config-heal-uneven-load) is set to.

A broker marked as permanently failed by Self-Balancing after being down for longer than `confluent.balancer.heal.broker.failure.threshold.ms`,
which then later recovers, is not considered in self-balancing operations for 24 hours (or until the controller changes).

The default is 1 hour (3600000 ms).

* Type: int
* Default: 3600000
* Importance: medium

<a id="sbc-config-throttle"></a>

### confluent.balancer.throttle.bytes.per.second

Specifies the maximum network bandwidth available for use by data balancing operations.
The value is specified in bytes/sec/broker. This is a [dynamic option](#sbc-kafka-configs-command),
meaning you can adjust the throttle while the cluster is running.

The default is 10 MB per second.

* Type: int
* Default: 10485760 (10 MB/sec)
* Importance: medium

#### How to choose a throttle setting

Self-Balancing uses the Apache Kafka® feature that lets you apply a throttle to replication traffic to limit the rate at
which data is moved when reassigning partitions as part of a rebalance. It’s important to understand that
when Self-Balancing is enabled, the configured throughput limit (`confluent.balancer.throttle.bytes.per.second`)
is always active, even if no rebalance is currently happening. This means any throughput limit you set is
continuously enforced on all replication traffic.

Because each cluster’s workload and hardware capabilities are different, it is difficult to
provide generally applicable guidance for choosing a throttle. Normal
replication traffic counts against this throttle, so in order for a rebalance to
make progress, the throttle must be larger than the highest inbound or outbound
replication rate of any broker in the cluster. These values are exposed as the
`kafka.server:type=BrokerTopicMetrics,name=ReplicationBytesInPerSec` and
`kafka.server:type=BrokerTopicMetrics,name=ReplicationBytesOutPerSec` broker metrics.

A safe method to choose a throttle is to start with a small multiple (3 to 5x)
of this maximum replication traffic, and then dynamically increase the throttle
in small increments, while monitoring cluster health. Produce
[request latency](https://docs.confluent.io/control-center/current/brokers.html#c3-brokers-request-latency-selector)
is a good metric to monitor, because it will increase if too much network
throughput is used for reassignment traffic or if too much data is being written
to disk as partitions are reassigned. This starting value is a conservative one,
and for some clusters, a much larger throttle may be safe.

Self-Balancing automatically removes the replication throttle when it is not reassigning
partitions, so as a cluster administrator you do not need to manually remove it.

<a id="sbc-confluent-balancer-disk-max-load"></a>

### confluent.balancer.disk.max.load

Specifies the maximum load for disk usage as a proportion the disk capacity (0-1.0)
that can be used when rebalancing is enabled.

The default is 85% of the log disk.

* Type: double
* Default: 0.85
* Importance: medium

If you receive an `OptimizationFailureException`, you might need to adjust the default value.
For details on how to correct this error, see [Rebalance fails with insufficient healthy cluster capacity](index.md#sbc-insufficient-healthy-capacity-troubleshoot) in Troubleshooting.

<a id="sbc-confluent-balancer-disk-max-replicas"></a>

### confluent.balancer.max.replicas

Specifies a maximum number of replicas allowed per broker.

The default value set in the example property files is 10,000 replicas. The recommended
value for this option is 4,000 replicas/partitions per broker.

* Type: long
* Default: 2147483647
* Importance: high

<a id="sbc-config-exclude-topic-names"></a>

### confluent.balancer.exclude.topic.names

Specifies topics that should not be moved by Self-Balancing. You can specify multiple topics in a comma-separated list.

- Removal operations always move topics, regardless of this setting.
- As a best practice, keep excluded topics to a minimum, and use this configuration
  only as necessary. Excluding too many topics can interfere with Self-Balancing.
  To learn more, see [Too many excluded topics cause problems with Self-Balancing](index.md#sbc-too-many-excluded-topics) in Troubleshooting.

For example, the following configuration would exclude the topics `pizza` and `popcorn`:

```bash
confluent.balancer.exclude.topic.names=pizza,popcorn
```

* Type: string
* Default: “”
* Importance: medium

<a id="sbc-config-exclude-topic-prefixes"></a>

### confluent.balancer.exclude.topic.prefixes

Specifies topics that should not be moved by Self-Balancing based on topic prefixes. You can specify multiple topic prefixes in a comma-separated list.

- Removal operations always move topics, regardless of this setting.
- As a best practice, keep excluded topics to a minimum, and use this configuration
  only as necessary. Excluding too many topics can interfere with Self-Balancing.
  To learn more, see [Too many excluded topics cause problems with Self-Balancing](index.md#sbc-too-many-excluded-topics) in Troubleshooting.

For example, the following configuration would exclude all topics prefixed with `pizza.sales.ny` and `pizza.sales.tx`:

```bash
confluent.balancer.exclude.topic.prefixes=pizza.sales.ny,pizza.sales.tx
```

* Type: string
* Default: “”
* Importance: medium

<a id="sbc-config-replication-factor"></a>

### confluent.balancer.topic.replication.factor

Specifies the replication factor for the topics the Self-Balancing uses to store internal state.

- For anything other than development testing, a value greater than 1 is recommended for `confluent.balancer.topic.replication.factor` to ensure availability.
- Replication factors can never be greater than the total number of brokers (regardless of Self-Balancing).
- When Self-Balancing is enabled, the value for `confluent.balancer.topic.replication.factor` and all other instances of `replication.factor` in `server.properties` must be set to a value greater than 1, but less than the total number of brokers in the cluster.
  If the value for a replication factor is equal to the total number of brokers deployed and a broker goes down or is removed, the replication factor will prevent rebalancing.

The default value is 3.

* Type: int
* Default: 3
* Importance: medium

## JBOD configurations

The following configurations enable and tune Self-Balancing balancing of data across the
disks within a broker that uses Just a Bunch of Disks (JBOD).
For an overview of the feature, see [Use JBOD with Self-Balancing in Confluent Platform](jbod.md#sbc-jbod).

<a id="sbc-config-logdirs-label"></a>

### confluent.telemetry.metrics.collector.volume.include.logDirs.label

Enables the broker to emit per-disk capacity metrics, which Self-Balancing reads to balance
data across disks.

Add this configuration to the properties file for every broker in the cluster so
that each broker emits per-disk capacity metrics. This configuration is not
[dynamic](#sbc-kafka-configs-command), so you must restart the brokers after
you change it.

* Type: boolean
* Default: false
* Importance: high

<a id="sbc-config-jbod-enable"></a>

### confluent.balancer.jbod.enable

Enables Self-Balancing to balance data across broker disks.

Add this configuration to the properties file for every controller in the cluster
and restart the controllers. You must also enable [confluent.telemetry.metrics.collector.volume.include.logDirs.label](#sbc-config-logdirs-label),
verify that [log.dirs](jbod.md#sbc-jbod-enable) is configured correctly for the
brokers, and enable Self-Balancing with [confluent.balancer.enable](#sbc-config-enable).

This configuration is not [dynamic](#sbc-kafka-configs-command), so you must
restart the controllers after you change it.

* Type: boolean
* Default: false
* Importance: high

<a id="sbc-config-intra-throttle"></a>

### confluent.balancer.intra.broker.throttle.bytes.per.second

Specifies the replication throttle limit for intra-broker movements, in
bytes/sec/broker. This is a [dynamic option](#sbc-kafka-configs-command),
meaning you can adjust the throttle while the cluster is running.

The default is 10 MB per second. Self-Balancing applies the throttle only to brokers
performing intra-broker movements and clears it after the movement completes.

* Type: int
* Default: 10485760 (10 MB/sec)
* Importance: medium

#### How to choose a throttle setting

Intra-broker movement consumes disk bandwidth, so set a throttle limit that
safeguards existing disk usage. The recommendation is to set the throttle
limit to half of the available bandwidth after subtracting the disk load:

```text
throttle <= ( min(read_bandwidth, write_bandwidth) - peak_current_load_per_disk ) / 2
```

The division by two accounts for the one-read plus one-write amplification.

<a id="sbc-config-intra-disk-max-load"></a>

### confluent.balancer.intra.broker.disk.max.load

Specifies the maximum load for an individual disk (per disk in a JBOD setup) as a
proportion of disk capacity. Valid values are between 0 and 1.

The default is 85% of the log disk.

* Type: double
* Default: 0.85
* Importance: medium

<a id="sbc-config-intra-movements-per-disk"></a>

### confluent.balancer.num.concurrent.intra.broker.partition.movements.per.disk

Specifies the maximum number of in-flight replicas per disk that Self-Balancing moves
during an intra-broker reassignment for any proposal.

* Type: int
* Default: 1
* Importance: medium

#### How to choose this value

The maximum number of replicas moved between disks is governed by
[num.replica.alter.log.dirs.threads](/platform/current/installation/configuration/broker-configs.html#num-replica-alter-log-dirs-threads),
where a single thread moves one replica. Disk parallelism should be at most twice
the number of threads available per disk (twice, because a disk acts as the source
for one replica and the destination for another). Set the number of threads based
on disk capability.

<a id="sbc-config-intra-disk-balance-threshold"></a>

### confluent.balancer.intra.broker.disk.balance.threshold

Specifies the maximum allowed extent of imbalance for disk utilization across the
disks within a broker. For example, `1.10` means the highest disk usage on a
broker should not exceed 1.10x the average disk utilization on that broker, and
the lowest disk usage should not fall below 0.9x the average. When an intra-broker
rebalance is triggered from the command line, Self-Balancing tries to bring the
broker’s disk usage into the range specified by this configuration.

* Type: double
* Default: 1.1
* Importance: medium

<a id="sbc-config-intra-disk-min-free"></a>

### confluent.balancer.intra.broker.disk.min.free.space.gb

Specifies the minimum amount of disk space, in GB, that must remain unused on each
individual disk in a JBOD setup. Valid values are between 0 and the disk size. The
balancer enforces the stricter of this configuration and
[confluent.balancer.intra.broker.disk.max.load](#sbc-config-intra-disk-max-load).

* Type: int
* Default: 0
* Importance: medium

<a id="sbc-kafka-configs-command"></a>

## Examples: Update broker configurations on the fly

To update most Self-Balancing settings, you must stop the brokers and shut down the cluster. However, the following cluster default settings are dynamic, meaning they can be performed while the cluster is running.

You can modify the values for these dynamic properties through Confluent Control Center or at the command line with the [kafka-configs](../../kafka/dynamic-config.md#kafka-dyn-broker-config) command and `--entity-default` flag, as shown in the following examples.

<a id="sbc-enable-disable-on-the-fly"></a>

### Enable or disable Self-Balancing

Use [confluent.balancer.enable](#sbc-config-enable) (turns Self-Balancing on or off).
Do not disable Self-Balancing while an add or remove broker operation is in progress; wait until the add or remove completes. Disabling Self-Balancing during either operation
fences off the broker from future partition distribution.

To turn Self-Balancing on:

```bash
kafka-configs --bootstrap-server <host:port> --entity-type brokers --entity-default --alter --add-config confluent.balancer.enable=true
```

To turn Self-Balancing off:

```bash
kafka-configs --bootstrap-server <host:port> --entity-type brokers --entity-default --alter --add-config confluent.balancer.enable=false
```

### Set trigger condition for rebalance

Use [confluent.balancer.heal.uneven.load.trigger](#sbc-config-heal-uneven-load) to rebalance only when brokers are added or removed, or
anytime for any uneven load.

Set Self-Balancing to rebalance on any uneven load (including a change in available brokers):

```bash
kafka-configs --bootstrap-server <host:port> --entity-type brokers --entity-default --alter --add-config confluent.balancer.heal.uneven.load.trigger=ANY_UNEVEN_LOAD
```

Set Self-Balancing to rebalance only when brokers are added or removed:

```bash
kafka-configs --bootstrap-server <host:port> --entity-type brokers --entity-default --alter --add-config confluent.balancer.heal.uneven.load.trigger=EMPTY_BROKER
```

### Set or remove a custom throttle

Use [confluent.balancer.throttle.bytes.per.second](#sbc-config-throttle) to set a custom throttle for maximum network bandwidth available for Self-Balancing or to remove a custom throttle.

To override the default throttle value for reassignment operations (10485760 or 10 MB/sec) with a custom, fixed value, you can use
the following commands. Dynamic configurations can also be changed through [API Reference for Confluent REST Proxy](../../kafka-rest/api.md#kafkarest-api).

```bash
kafka-configs --bootstrap-server <host:port> --entity-type brokers --entity-default --alter --add-config confluent.balancer.throttle.bytes.per.second=70485760
```

To disable a custom throttle:

```bash
kafka-configs --bootstrap-server <host:port> --entity-type brokers --entity-default --alter --delete-config confluent.balancer.throttle.bytes.per.second
```

<a id="sbc-internal-topics"></a>

## Self-Balancing internal topics

Self-Balancing creates the following system or internal topic, which it uses to manage cluster auto-balancing.

- `_confluent_balancer_api_state`: Stores state information used by the Self-Balancing APIs.

Additionally, Self-Balancing requires metrics on cluster performance from the [Configure Telemetry Reporter for Confluent Platform](../../health-plus/telemetry.md#telemetry-reporter), which is enabled by default
(`confluent.reporters.telemetry.auto.enable = true`) when Self-Balancing is enabled.
Metrics needed specifically for Self-Balancing on processing loads, message queue size,
response times, replication data, and more are sent to the internal topic
`_confluent-telemetry-metrics`. Self-Balancing [reads from this internal topic](../../health-plus/enable-health-plus.md#telemetry-reporter-internal-topic-for-data-collection) and uses the metrics in
its balancing algorithm. With Self-Balancing running, `_confluent-telemetry-metrics`
can collect a large amount of data in a short time. The default retention period
for internal topics is 3 days, however you can use `kafka-topics --config` to
modify this. A retention period of 1 day is typically sufficient for this topic
with regards to Self-Balancing.

By convention, Confluent Platform system topics are prefixed with an underscore. The Self-Balancing topic is prefixed with `_confluent_balancer_`. You can get a list of existing topics (user-created and system topics) as follows:

```bash
kafka-topics --list --bootstrap-server localhost:9092
```

Get detailed information on all topics or a specific topic with the `--describe` option:

```bash
kafka-topics --describe --bootstrap-server localhost:9092
```

```bash
kafka-topics --describe --topic <topic> --bootstrap-server localhost:9092
```

For example, run `kafka-topics --describe` on the `_confluent_balancer_api_state` topic to view details.

<a id="sbc-configs-c3"></a>

## Required configurations for Control Center

Self-Balancing requires embedded Confluent REST Proxy to communicate with Confluent Control Center monitoring.

### Configure REST endpoints in the Control Center properties file

If you want to use Control Center with Self-Balancing for [Configuration and monitoring](index.md#sbc-config-and-monitor), you
must configure the Control Center cluster with REST endpoints to enable HTTP servers
on the brokers. If this is not configured properly for all brokers, Self-Balancing will
not be accessible from Confluent Control Center.

In the appropriate Control Center properties file, use `confluent.controlcenter.streams.cprest.url`
to define the REST endpoints for `controlcenter.cluster`. The default is `http://localhost:8090`,
as shown in the following example.

```bash
# Kafka REST endpoint URL
confluent.controlcenter.streams.cprest.url="http://localhost:8090"
```

Identify the associated URL for each broker. If you have multiple brokers in the cluster, use a comma-separated list.

#### SEE ALSO
- An example of configuring this on a five-broker cluster in [Configure Control Center with REST endpoints and advertised listeners](sbc-tutorial.md#c3-cprest-config-tutorial) in the Self-Balancing Tutorial
- `confluent.controlcenter.streams.cprest.url` in the [Control Center Configuration Reference](https://docs.confluent.io/control-center/current/installation/configuration.html#c3-connect-cluster-setting)
- [Self-Balancing options do not show up on Control Center](index.md#sbc-not-on-c3-troubleshoot) in Troubleshooting

<a id="sbc-rest-endpoint-configs-secure-setup"></a>

### Configure authentication for REST endpoints on Kafka brokers (secure setup)

To run Self-Balancing in a secure setup, you must configure authentication for REST endpoints in each of the Kafka broker `server.properties` files.
If the Kafka broker files are missing these configurations,
Control Center cannot access Self-Balancing in a secure setup.

At a minimum, you will need the following configurations.

```bash
# EmbeddedKafkaRest: HTTP Auth Configuration
kafka.rest.kafka.rest.resource.extension.class=io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension
kafka.rest.rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler
```

Here is an example of an MDS client configuration for Kafka RBAC in a broker `server.properties` file.

```bash
# EmbeddedKafkaRest: Kafka Client Configuration
kafka.rest.bootstrap.servers=<host:port>,<host:port>,<host:port>
kafka.rest.client.security.protocol=SASL_PLAINTEXT

# EmbeddedKafkaRest: HTTP Auth Configuration
kafka.rest.kafka.rest.resource.extension.class=io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension
kafka.rest.rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler
kafka.rest.public.key.path=<rbac_enabled_public_pem_path>

# EmbeddedKafkaRest: MDS Client configuration
kafka.rest.confluent.metadata.bootstrap.server.urls=<host:port>,<host:port>,<host:port>
kafka.rest.ssl.truststore.location=<truststore_location>
kafka.rest.ssl.truststore.password=<password>
kafka.rest.confluent.metadata.http.auth.credentials.provider=BASIC
kafka.rest.confluent.metadata.basic.auth.user.info=<user:password>
kafka.rest.confluent.metadata.server.urls.max.age.ms=60000
kafka.rest.client.confluent.metadata.server.urls.max.age.ms=60000
```

#### SEE ALSO
- [Security considerations](index.md#sbc-security-considerations)
- [Configure Security for the Admin REST APIs for Confluent Server on Confluent Platform](../../kafka-rest/production-deployment/confluent-server/security.md#confluent-server-rest-security)
- [Deploy Secure Standalone REST Proxy in Confluent Platform](../../kafka-rest/production-deployment/rest-proxy/security.md#kafkarest-security)
- [Scripted Confluent Platform Demo](../../tutorials/cp-demo/index.md#cp-demo), On-Prem Tutorial, Security section provides examples of different types of configurations:

<a id="sbc-kafka-rebalance-cluster"></a>

## Monitoring the balancer with kafka-rebalance-cluster

Confluent Platform provides [APIs](../../kafka-rest/index.md#kafkarest-intro) to gain visibility into the
status of Self-Balancing and its operations.
You can use the APIs to retrieve these statuses programmatically.

Also, a command-line tool that installs with Confluent Platform, [kafka-rebalance-cluster](../../tools/cli-reference.md#kafka-rebalance-cluster-tool),
uses the APIs to report the same information on the command-line.

### Get the balancer status

The balancer status API provides visibility into the state of the Self-Balancing component itself.

Use `kafka-rebalance-cluster` with the `--status` flag to query the status of the balancer.
Following are example status messages you may receive as output.

```bash
bin/kafka-rebalance-cluster.sh  --bootstrap-server localhost:9089 --status
Balancer status: ENABLED
```

```bash
bin/kafka-rebalance-cluster.sh  --bootstrap-server localhost:9089 --status
Balancer status: ERROR
Error description: SBC configured with multiple log directories
```

### Get the workload optimization status (AnyUnevenLoad)

If the [balancer is set to rebalance](#sbc-config-heal-uneven-load) on
`ANY_UNEVEN_LOAD`, it will automatically detect and act upon uneven
distribution of workloads within a cluster. The `even-cluster-load` API provides
visibility into whether a goal violation for workload distribution has been met
and what Self-Balancing is currently doing about it. It provides more context in case
balancing fails due to some internal error or user intervention.

Run `kafka-rebalance-cluster` with the `--describe` flag to get the
fine-grained status of these balancing optimizations.

```bash
bin/kafka-rebalance-cluster.sh  --bootstrap-server localhost:9089 --describe
```

Following are example outputs for this command in various scenarios.

- When the [balancer](#sbc-config-heal-uneven-load) is set to `EMPTY_BROKER` (not `ANY_UNEVEN_LOAD`):
  ```bash
  Uneven load balance status:
   Current: DISABLED
  ```
- After startup, but before any uneven load balancing operation has run:
  ```bash
  Uneven load balance status:
   Current: STARTING
  ```
- When a goal violation detection has succeeded:
  ```bash
  Current: BALANCED
  Last Update Time: 2021-02-26_23:09:23 UTC
  Previous: BALANCED
  Last Update Time: 2021-02-26_23:07:23 UTC
  ```
- During rebalancing:
  ```bash
  Current: BALANCING
  Last Update Time: 2021-02-26_23:25:23 UTC
  Previous: BALANCED
  Last Update Time: 2021-02-26_23:19:23 UTC
  ```

### Balance data across broker disks (JBOD)

In a JBOD setup, you can use `kafka-rebalance-cluster`
to balance data across the disks within a broker. The following flags trigger
intra-broker rebalancing:

- `--intra-broker-rebalance-dry-run`: Compute and display the rebalance plan
  without moving any data.
- `--intra-broker-rebalance`: Start moving replicas between disks.

You must enable JBOD support before using this feature. For prerequisites,
configuration, command examples, and limitations, see [Use JBOD with Self-Balancing in Confluent Platform](jbod.md#sbc-jbod).

<a id="sbc-command-remove-brokers"></a>

## kafka-remove-brokers

Confluent Enterprise releases include the `kafka-remove-brokers` command-line tool, which is available in the
`bin` directory of your Confluent Platform installation.

Self-Balancing must be enabled for this command to work.

Note the following about removing brokers:

- In cases where metrics collection is not yet complete, an attempt to remove a broker fails almost immediately
  due to insufficient metrics for generating a rebalancing plan. If broker removal fails, wait for 30 minutes, then retry. Once Self-Balancing has
  initialized and had time to collect metrics, the operation should succeed. To learn more, see [Broker removal attempt fails during Self-Balancing initialization](index.md#sbc-remove-broker-troubleshoot) in [Troubleshooting](index.md#sbc-troubleshooting).
- You can also remove a broker with [Control Center](https://docs.confluent.io/control-center/current/clusters.html#c3-sbc-remove-broker).
- Do not [disable Self-balancing while an add or remove broker operation is in progress](#sbc-enable-disable-on-the-fly); wait until the add or remove completes.
  Disabling Self-Balancing during either operation fences off the broker from future partition distribution.
- Self-Balancing Clusters supports (KRaft) mode. For more information, see [KRaft Overview for Confluent Platform](../../kafka-metadata/kraft.md#kraft-overview). If you are using
  [Self-balancing with Confluent for Kubernetes](https://docs.confluent.io/operator/current/co-scale-cluster.html#scale-cp-clusters-and-balance-data),
  and using CFK to shut down the broker, there is no further action needed. However, if you are running Confluent Platform outside of CFK on a KRaft supported version of Confluent Platform,
  and separately using the `kafka-remove-brokers` tool, you must take an additional action to call the `UnregisterBroker` flag through `kafka-cluster`
  after the broker removal is complete. If you call [kafka-remove-brokers](#sbc-command-remove-brokers) with `shouldShutdown=false`, you must manually remove the broker; if you
  set `shouldShutdown=true`, Self-Balancing will shut down the broker for you. In either case, with KRaft mode Confluent Platform deployments,
  you must call `UnregisterBroker` after broker shutdown is complete to cleanup lingering metadata within the controllers.
  This will cleanly remove all traces of the broker on the cluster.

### kafka-remove-brokers flags

The `kafka-remove-brokers` tool provides the following required and optional flags.

`--bootstrap-server`
: The connection string for the cluster’s broker(s) in the form `host:port`.
  You can specify multiple URLs, separated by commas, to allow for failover
  should a broker node go down. (REQUIRED)
  <br/>
  * Type: string
  * Default: empty string
  * Importance: high

`--delete`
: Remove one broker from the cluster.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`--describe`
: Describe the status of broker removal.
  <br/>
  * Type: string
  * Default: “”
  * Importance: medium

`--broker-id`
: The ID of the broker to remove. (REQUIRED)
  <br/>
  * Type: string
  * Default: “”
  * Importance: low

`--broker-ids`
: The IDs of the brokers to remove.
  <br/>
  * Type: string
  * Default: “”
  * Importance: low

`--command-config`
: Specifies a property file containing configurations to be passed to the Admin Client. This
  option is used only with the `--bootstrap-server` option.
  <br/>
  * Type: string
  * Default: “”
  * Importance: low

`--no-shutdown`
: Prevents shutdown of brokers as part of the removal operation. By default, the brokers are shut down as part of broker removal.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`--version`
: Display Kafka version.
  <br/>
  * Type: string
  * Default: “”
  * Importance: low

### kafka-remove-brokers examples

- The following command removes broker 1 from a cluster of five brokers and
  redistributes its partitions across the remaining brokers.
  ```bash
  ./bin/kafka-remove-brokers --bootstrap-server localhost:9092,localhost:9093,localhost:9094,localhost:9095,localhost:9096 --broker-id 1 --delete 1>&2 | grep -v SLF4J
  ```
- The following command provides the status of this rebalance operation. The `--describe` flag is substituted for `--delete`.
  ```bash
  ./bin/kafka-remove-brokers --bootstrap-server localhost:9092,localhost:9093,localhost:9094,localhost:9095,localhost:9096 --broker-id 1 --describe 1>&2 | grep -v SLF4J
  ```

<a id="sbc-broker-removal-phases"></a>

## Broker removal phases

Understanding the broker removal process can be useful for monitoring and troubleshooting Self-Balancing.
To accomplish broker removal, Self-Balancing performs the following consecutive tasks:

1. Validates that broker metrics are available on the controller (lead broker) running Self-Balancing (a type of dry run).
2. Places replica exclusions on the brokers to be removed. This prevents new replicas from being placed on those brokers during the removal process.
3. Computes the partition reassignment (rebalancing) plan.
4. Executes the plan to reassign partitions, and moves topic data.
5. Shuts down the broker(s).
6. Removes the exclusions.

The last two phases (broker shutdown and remove exclusions) are optional. These are performed only if `shouldShutdown=true`,
which is the default.

Broker removal can fail if attempted while Self-Balancing is still initializing, as described in these scenarios:

**Scenario 1:** Any attempt to remove a broker during initialization will fail if there are insufficient metrics
(at phase 1). The solution is to wait for Self-Balancing to initialize (about 30 minutes), and retry the broker removal.

**Scenario 2:** An attempt to remove the controller during initialization can fail at phase 3.
In this scenario, Self-Balancing is running long enough to collect sufficient metrics to compute a plan
on the lead broker, so it passes phases 1 and 2 and shuts down the broker you want to remove.
But Self-Balancing metrics collection has not progressed long enough to provide Self-Balancing metrics on all the
other brokers, one of which will be the failover controller. This causes the remove operation to
stall out and, since the controller was shut down (phase 2) it will not be available on Control Center.
The solution here is similar to scenario 1; wait and retry the broker removal, but from the command line.

For full details, see [Broker removal attempt fails during Self-Balancing initialization](index.md#sbc-remove-broker-troubleshoot).

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

## Self-Balancing initialization

Self-Balancing requires about 30 minutes to initialize, whether you enable it on a
running cluster or start a cluster with Self-Balancing already enabled. Initialization
consists of metrics collection by the lead broker across all brokers in the
cluster. Self-Balancing must have sufficient metrics on the cluster in order to compute
rebalancing plans.

Therefore, metrics collection is a prerequisite to performing Self-Balancing tasks. If
broker removal is attempted during the initialization process, the action will
fail. The solution is to retry the broker removal after Self-Balancing is initialized.
You may have to do this from the command line if the broker you tried to remove
is the controller because it may not be available on Control Center. For full details,
see [Broker removal phases](#sbc-broker-removal-phases) and [Broker removal attempt fails during Self-Balancing initialization](index.md#sbc-remove-broker-troubleshoot).

<a id="sbc-broker-removal-priority"></a>

## Broker removal task priority

Self-Balancing elevates the relative priority of a broker removal request above other tasks like adding a broker or performing a normal [rebalance](index.md#sbc-rebalance-trigger-conditions):

- Priority 1: Remove broker
- Priority 2: Add broker
- Priority 3: Normal rebalance

An ongoing broker removal request takes priority over a follow-on request. Self-Balancing will reject a new “remove broker” request while another broker removal task is in progress.

If a new “add broker” request is received while another “add broker” task is in progress, Self-Balancing will merge the new request with the in-progress task.

## Related content

- [Tutorial: Add and Remove Brokers with Self-Balancing in Confluent Platform](sbc-tutorial.md#sbc-tutorial)
- [Use kafka-configs to changing broker configurations dynamically](../../kafka/dynamic-config.md#kafka-dyn-broker-config)
- [Confluent REST Proxy for Apache Kafka on Confluent Platform](../../kafka-rest/index.md#kafkarest-intro)
- [API Reference for Confluent REST Proxy](../../kafka-rest/api.md#kafkarest-api)
- [Work with Self-Balancing Clusters](https://docs.confluent.io/control-center/current/clusters.html#c3-sbc) in the Control Center guide