...

You can then use the kafka-configs.sh tool to described the cordoned log directories:
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type brokers --entity-name 0
Dynamic configs for broker 0 are:
  cordoned.log.dirs=/tmp/kraft-broker0-logs0 sensitive=false synonyms={DYNAMIC_BROKER_CONFIG:cordoned.log.dirs=/tmp/kraft-broker0-logs0}

You can use the same mechanisms to uncordon log directories. Either update the broker's properties file and restart the broker, or use the kafka-configs.sh tool to update the cordoned.log.dirs configuration.

When the configuration is set on a broker, the broker will include the Uuids of cordoned log directories when it registers to the controller and also include them in each heartbeat request it sends to the controller to reflect potential dynamic updates.

...

{
  "apiKey": 63,
  "type": "request",
  "listeners": ["controller"],
  "name": "BrokerHeartbeatRequest",
  "validVersions": "0-1",
  "flexibleVersions": "10+",
  "fields": [
     ...
     { "name": "CordonedLogDirs", "type":  "[]uuid", "versions": "1+", "taggedVersions": "2+", 
       "tag": "1", "about": "Log directories that are cordoned." }
  ]
}

• BrokerHeartbeatResponse v1 adds no new fields
  

Metadata Records

Updating these metadata records ensures that the cordoned log directories of each broker are persisted in the metadata log. So if a broker is offline, its cordoned log directories are still available to the controller when assigning partitions.

• BrokerRegistrationChangeRecord v3 adds the CordonedLogDirs field

...

{
  "apiKey": 35,
  "type": "response",
  "name": "DescribeLogDirsResponse",
  // Starting in version 1, on quota violation, brokers send out responses before throttling.
  "validVersions": "0-5",
  // Version 2 is the first flexible version.
  // Version 3 adds the top-level ErrorCode field
  // Version 4 adds the TotalBytes and UsableBytes fields
  // Version 5 adds the IsCordoned fields
  "flexibleVersions": "2+",
  "fields": [
      ...
      { "name": "IsCordoned", "type": "bool", "versions": "5+", "ignorable": true, "default": false,
        "about": "True if this log directory is cordoned."
      }
    ]}
  ]
}

Admin API

When describing log directories via the Admin.describeLogDirs() method, you get LogDirDescription objects. A new method will be added to LogDirDescription to tell whether a the log directory is cordoned:

/**
 * A description of a log directory on a particular broker.
 */
public class LogDirDescription {

    ...

    /**
     * Whether this log directory is cordoned or not.
     */
    public boolean isCordoned() {
       ...
    }
}

The output of the LogDirsCommand tool will also be updated to show if a log directory is cordoned or not.

Compatibility, Deprecation, and Migration Plan

...