<a id="cordon-broker"></a>

# Cordon Brokers and Log Directories in Confluent Platform

Cordoning marks a Apache Kafka® broker or an individual log directory as off-limits for
new partition placement while the existing partitions on that broker or log
directory keep serving reads and writes. Cordoning is the first step when you
decommission a broker or a log directory, because it prevents the controller
from assigning new partitions to a resource that you are about to remove.

Cordoning works like `kubectl cordon`. The broker stays in the cluster and
continues to serve its current partitions, but the controller does not place
any new partitions on a cordoned broker or log directory.

If your cluster uses Self-Balancing Clusters or Confluent Auto Data Balancer, do not use cordoning. These
features manage replica placement and do not account for cordoned brokers or
log directories, so they can place replicas on cordoned resources. For more
information, see [Limitations and known issues](../clusters/sbc/index.md#sbc-limitations) and [Auto Data Balancing in Confluent Platform](../clusters/rebalancer/index.md#rebalancer-overview). To
remove a broker from a cluster that uses Self-Balancing Clusters, see
[Remove a broker](../clusters/sbc/sbc-tutorial.md#sbc-remove-broker).

#### IMPORTANT
Broker and log directory cordoning requires the Confluent Platform 8.3 metadata version or
later. Until the metadata version supports cordoning, the
`cordoned.log.dirs` configuration has no effect, and an attempt to set it
dynamically is rejected with an `InvalidConfigurationException`. To
finalize the metadata version, see [Steps for upgrading to 8.3.x](../installation/upgrade.md#upgrade-kraft-cluster).

`cordoned.log.dirs` is the broker configuration that controls cordoning, and
it takes effect only at the Confluent Platform 8.3 metadata version or later. Set
`cordoned.log.dirs` to a comma-separated list of log directory paths to
cordon, or set it to `*` to cordon all log directories on the broker. Each
path that you list in `cordoned.log.dirs` must also be configured in the
broker’s `log.dirs` or `log.dir` setting. For the broker configuration
reference, see [Kafka Broker and Controller Configuration Reference for Confluent Platform](../installation/configuration/broker-configs.md#cp-config-brokers).

Use the `kafka-configs` tool to set `cordoned.log.dirs` dynamically through
the Admin API, or set it statically in the broker’s `server.properties` file,
where it takes effect on the next broker restart. You can check whether a log
directory is cordoned through the `LogDirDescription.isCordoned()` method of
the Admin API.

When a broker or log directory is cordoned, the controller excludes it from
placement of the replicas for new partitions. If the remaining brokers and log
directories cannot satisfy the replication factor, topic and partition creation
fails with an `InvalidReplicationFactorException`.

## Decommission a broker

To decommission a broker, cordon it, move its partitions to other brokers, shut
it down, and then unregister it.

1. Cordon the broker by setting `cordoned.log.dirs` to `*` through the Admin
   API. The following example cordons broker `1`:
   ```bash
   bin/kafka-configs --bootstrap-server localhost:9092 --alter \
     --add-config cordoned.log.dirs="*" --entity-type brokers --entity-name 1
   ```

   The command returns the following output:
   ```text
   Completed updating config for broker 1.
   ```
2. Reassign all partitions from the cordoned broker to the other brokers in the
   cluster. The partition reassignment tool cannot generate a reassignment plan
   for decommissioning automatically, so you must create a plan that moves
   every replica hosted on the broker to the remaining brokers. For the
   reassignment procedure, see [Scaling the cluster (adding a node to a Kafka cluster)](post-deployment.md#scaling-the-cluster).

   Move partitions only to brokers and log directories that are not cordoned. A
   move that targets a cordoned log directory fails with an
   `InvalidReplicaAssignmentException` and the message
   `Log directory <dir> is cordoned`.
3. After the reassignment finishes, shut down the broker.
4. Unregister the broker to remove it from the cluster. The following example
   unregisters broker `1`:
   ```bash
   bin/kafka-cluster unregister --bootstrap-server localhost:9092 --id 1
   ```

## Decommission a log directory

To decommission a single log directory on a broker, cordon the directory, move
its partitions elsewhere, shut down the broker, uncordon the directory, remove
it from the broker configuration, and then restart the broker.

1. Cordon the log directory by setting `cordoned.log.dirs` to the directory
   path through the Admin API. The following example cordons `/data/dir1` on
   broker `1`:
   ```bash
   bin/kafka-configs --bootstrap-server localhost:9092 --alter \
     --add-config cordoned.log.dirs=/data/dir1 --entity-type brokers --entity-name 1
   ```

   The command returns the following output:
   ```text
   Completed updating config for broker 1.
   ```
2. Reassign all partitions from the cordoned log directory to other log
   directories or brokers in the cluster. The partition reassignment tool
   cannot generate a reassignment plan for decommissioning a log directory
   automatically, so you must create a plan that moves every replica hosted on
   the log directory. For the reassignment procedure, see
   [Scaling the cluster (adding a node to a Kafka cluster)](post-deployment.md#scaling-the-cluster).

   Choose target log directories that are not cordoned. A move that targets a
   cordoned log directory fails with an `InvalidReplicaAssignmentException` and
   the message `Log directory <dir> is cordoned`.
3. After the reassignment finishes, shut down the broker.
4. Uncordon the log directory. Because the broker that hosts the directory is
   offline, use the `--bootstrap-controller` option to reach the controller
   directly. The following example uncordons the log directory on broker `1`:
   ```bash
   bin/kafka-configs --bootstrap-controller localhost:9093 --alter \
     --delete-config cordoned.log.dirs --entity-type brokers --entity-name 1
   ```

   The command returns the following output:
   ```text
   Completed updating config for broker 1.
   ```
5. Remove the decommissioned log directory from the `log.dirs` broker
   configuration. For example, if the broker configuration contained the
   following setting:
   ```properties
   log.dirs=/data/dir1,/data/dir2
   ```

   Update the setting to remove the decommissioned directory:
   ```properties
   log.dirs=/data/dir2
   ```
6. Restart the broker.

## Related content

* [KRaft Overview for Confluent Platform](../kafka-metadata/kraft.md#kraft-overview)
* [KRaft Configuration for Confluent Platform](../kafka-metadata/config-kraft.md#configure-kraft)
* [Scaling the cluster (adding a node to a Kafka cluster)](post-deployment.md#scaling-the-cluster)
* [Kafka Broker and Controller Configuration Reference for Confluent Platform](../installation/configuration/broker-configs.md#cp-config-brokers)
* Apache Wiki: [KIP-1066: Mechanism to cordon brokers and log directories](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1066%3A+Mechanism+to+cordon+brokers+and+log+directories)