...

The controller uses the ClusterDescriber interface to let the replica placer determine the usable brokers and the default log directory partitions should be placed on. This interface has 2 methods whose behavior will be updated:
- usableBrokers(): This method will only return brokers that have at least one log directories directory not cordoned.
- defaultDir(): This method will not return the Uuid of a log directory that is not cordoned.

Error handling

CreateTopics/CreatePartitions requests

When creating a new topic or partitions, if not enough brokers are usable to satisfy the requested replication factor, the INVALID_REPLICATION_FACTOR error will be returned. This error is already returned today is not enough brokers are registered. The error message will be adjusted to mention this could also be due to brokers not having uncordoned log directories as the current error message only mentioned the number of registered brokers.

AlterPartitions request

When reassigning partitions between brokers, if the request places partitions on brokers with no uncordoned log directories, the INELIGIBLE_REPLICA will be returned. Currently the controller also logs an INFO message with the reason. This log will be updated to provide the correct reason in case the request is rejected because of cordoned log directories.

AlterReplicaLogDirs request

When reassigning partitions onto a cordoned log directory, the KAFKA_STORAGE_ERROR error will be returned with a message indicating the target directory is cordoned. This error is already returned today if the target directory is offline.

Public Interfaces

This KIP introduces a new broker configuration, new API versions and new metadata records versions.

...

• BrokerRegistrationResponse v4 adds no new fields
  
  
• BrokerHeartbeatRequest v1 adds the CordonedLogDirs flexible field

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

...

{
  "apiKey": 17,
  "type": "metadata",
  "name": "BrokerRegistrationChangeRecord",
  "validVersions": "0-3",
  "flexibleVersions": "0+",
  "fields": [
     ...
     { "name": "CordonedLogDirs", "type":  "[]uuid", "versions":  "3+", "taggedVersions": "3+", 
       "tag": "3", "about": "Log directories that are cordoned." }
  ]
}

• RegisterBrokerRecord v4 adds the CordonedLogDirs field

{
  "apiKey": 0,
  "type": "metadata",
  "name": "RegisterBrokerRecord",
  "validVersions": "0-4",
  "flexibleVersions": "0+",
  "fields": [
     ...
     { "name": "CordonedLogDirs", "type":  "[]uuid", "versions":  "4+", "taggedVersions": "4+", 
       "tag": "1", "about": "Log directories that are cordoned." }
  ]
}

...

• KIP-660: Pluggable ReplicaPlacer: This proposed making the ReplicaPlacer pluggable and allow users to implement their own. The main issue was that in order to make an intelligent placement, the placer would need to have a full view of the cluster which is potentially very large. Implementations would potentially be complex and also need to be fast to not hang the controller thread. For these reasons, this proposal was rejected.