Page History

...

If log directory that holds the cluster metadata topic is configured separately to a different path — using metadata.log.dir — — then this log directory is does not get a UUID assigned. 

Reserved UUIDs

The following UUIDs are excluded from the random pool when generating a log directory UUID:

• UUID.UnknownDir – new Uuid(0L, 0L) – used to identify new or unknown assignments.
• UUID.OfflineDir - new Uuid(0L, 1L) – used to represent unspecified offline directories.

Metadata records

RegisterBrokerRecord and BrokerRegistrationChangeRecord will both have two new fields:

{ "name": "OnlineLogDirs", "type":  "[]uuid", "versions":  "3+", "taggedVersions": "3+", "tag": "0",
  "about": "Log directories configured in this broker which are available." },
{ "name": "OfflineLogDirs", "type":  "bool", "versions":  "3+",
  "about": "Whether any log directories configured in this broker are not available." }

...

{ "name": "Replicas", "type":  "[]int32", "versions":  "0", "entityType": "brokerId",
  "about": "The replicas of this partition, sorted by preferred order." },
(...)
{ "name": "Assignment", "type": "[]ReplicaAssignment", "versions": "1+",
  "about": "The replicas of this partition, sorted by preferred order.", "fields": [
  { "name": "Broker", "type": "int32", "versions": "1+", "entityType": "brokerId",
    "about": "The broker ID hosting the replica." },
  { "name": "Directory", "type": "uuid", "versions": "1+", "taggedVersions": "+1", "tag": "0",
    "about": "The log directory hosting the replica" }
]}

The new Directory field is a tagged field, which won't be used when a single log directory is registered, avoiding the need to grow the metadata record size for non JBOD-configured clusters.

Although not explicitly specified in the schema, the default value for Directory is Uuid.UnknownDir, as that's the default default value for UUID types.

...

{ "name": "OnlineLogDirs", "type":  "[]uuid", "versions":  "2+", "taggedVersions": "2+", "tag": "0",
  "about": "Log directories configured in this broker which are available." },
{ "name": "OfflineLogDirs", "type":  "bool", "versions":  "2+",
  "about": "Whether any log directories configured in this broker are not available." }

...

Storage format command

Configuration

The following configuration option is introduced

Storage format command

The formatThe format subcommand will be updated to ensure each log directory has an assigned UUID and it will persist a new property directory.id in the meta.properties  file when multiple log.dirs are configured. The value is base64 encoded, like the cluster UUID.

...

...

When the broker starts up and initializes LogManager, if multiple log.dirs are configured, it will load the UUID for each log directory (directory.id ) by reading the meta.properties file at the root of each of them.

• If there are any two log directories with the same UUID, the broker Broker will fail at startup
• If there are any meta.properties files missing directory.id, a new UUID is generated, and assigned to that log directory by updating the meta.properties file.

The set of all loaded log directory UUIDs is sent along in the broker registration request to the controller as the OnlineLogDirs field. If any configured log directories is unavailable, OfflineLogDirs is set to true.

Metadata caching

Currently, Replicas are considered offline if the hosting broker is offline, or if the hosting broker's registration flags offline log directories and the replica references none of the registered online log directories.

Handling log directory failures

When one or more log directories become offline, the broker will communicate this change using the new field OfflineLogDirs  in the BrokerHeartbeat  request — indicating the UUIDs of the new offline log directories. The UUIDs for the newly failed log directories are included in the BrokerHeartbeat  request until the broker sees them correctly identified as offline in a BrokerRegistrationChangeRecord.

. Additionally, replicas will also be considered offline if the replica references a log directory UUID (in the new field partitionRecord.Assignment.Directory) that is not present in the hosting Broker's latest registration under OnlineLogDirs and either:

• the hosting broker's latest registration indicates multiple online log directories. i.e. brokerRegistration.OnlineLogDirs.length > 1
• the hosting broker's latest registration indicates that there are offline directories. i.e. brokerRegistration.OfflineLogDirs == true

If neither of the above conditions are true, we assume that there is only one log directory configured, the broker is not configured with multiple log directories, replicas all live in the same directory and neither log directory assignments nor log directory failures shall be communicated to the Controller. 

Handling log directory failures

When multiple log directories are configured, and some (but not all) of them become offline, the broker will communicate this change using the new field OfflineLogDirs  in the BrokerHeartbeat  request — indicating the UUIDs of the new offline log directories. The UUIDs for the newly failed log directories are included in the BrokerHeartbeat  request until the broker receives a successful response. If the Broker is configured with a single log directory, this field isn't used, as the current behavior of the broker is to shutdown when no log directories are online.

If there are any queued partition-to-directory assignments to send to the controller, those that are respective to any of the newly failed log directories (i.e. assignments that are either into or out-of these directories) are prioritized and sent first. The broker retries these until it receives a successful reply, which conveys that the metadata change has been successfully persisted. This ensures that the Controller is in sync with regards to partition-to-directory assignments and can reliably determine which partitions need leadership and ISR update.

If the Broker repeatedly cannot communicate fails to communicate a log directory failure after a configurable amount of time — log.dir.failure.timeout.ms — and it is the leader for any replicas in the failed log directory the broker will shutdown, as that is the only other way to guarantee that the controller will elect a new leader for those partitionsBecause the broker is proactive in communicating any log directory assignment changes to the controller, the metadata should be up to date and correct when the controller is notified of a failed log directory. However, the consequences of some partition assignment being incorrect – due to some error or race condition - can be quite damaging, as the controller might not know to update leadership for that partition, leaving it unavailable for an indefinite amount of time. So, as a fallback mechanism, when handling a runtime directory failure, the broker must verify the assignments for the newly failed partitions against the latest metadata, and for any incorrect assignments, the broker will use AlterReplicaLogDirs to assign them to UUID.OfflineDir so that the controller can update leadership and ISR.

Replica management

When configured with multiple log.dirs, as the broker catches broker catches up with metadata, and sees the partitions which it should be hosting, it will check the associated log directory UUID for each partition (partitionRecord.Assignment.Directory).

• If the partition is not assigned to a log directory (refers to Uuid.UnknownDir)
  • If the partition already exists, the broker uses the new RPC — AssignReplicasToDirs — to notify the controller to change the metadata assignment to the actual log directory.
  • If the partition does not exist, the broker selects a log directory and uses the new RPC — AssignReplicasToDirs — to notify the controller to create the metadata assignment to the actual log directory.
• If the partition is assigned to an online log directory
  • If the partition does not exist it is created in the indicated log directory.
  • If the partition already exists in the indicated log directory and no future replica exists, then no action is taken.
  • If the partition already exists in the indicated log directory, and there is a future replica in another log directory, then the broker starts the process to replicate the current replica to the future replica.
  • If the partition already exists in another online log directory and is a future replica in the log directory indicated by the metadata, the broker will replace the current replica with the future replica after making sure that the future replica is fully caught up with the current replica.
  • If the partition already exists in another online log directory, the broker uses the new RPC — AssignReplicasToDirs — to the controller to change the metadata assignment to the actual log directory. The partition might have been moved to a different log directory whilst the broker was offline. 
• If the partition is assigned to an unknown log directory or refers to Uuid.OfflineDir
  • If there are offline log directories, no action is taken — the assignment refers to a a log directory which may be offline, we don't want to fill the remaining online log directories with replicas that existed in the offline ones.
  • If there are no offline directories, the broker selects a log directory and uses the new RPC — AssignReplicasToDirs — to notify the controller to create the metadata assignment to the actual log directorylog directory.

If instead, a single entry is configured under log.dirs or log.dir, then the Assignment.Directory field is always ignored, and AssignReplicasToDirs is never sent.

If the broker is configured with multiple log directories it remains FENCED until it can verify that all partitions are assigned to the correct log directories in the cluster metadata. This excludes the log directory that hosts the cluster metadata topic, if it is configured separately to a different path — using metadata.log.dir.

...

Once the broker receives confirmation of the metadata change – indicated by fetching cluster metadata, and noticing a new PartitionChangeRecord which reflects the log directory assignment change for the moving replica – a successful response to AssignReplicasToDirs – then it will:

1. Block appends to the main (old) replica and waits for the future replica to fully catch up once again.
2. Makes the switch, promoting the future replica to main replica and cleaning up the old one.

...

The diagram below illustrates the sequence of steps involved in moving a replica between log directories.Image Removed

Image Added

In the diagram above, notice that if dir1 fails after the AssignReplicasToDirs RPC is sent, but before the future replica is promoted, then the controller will not know to update leadership and ISR for the partition. This is why it is important to have the fallback mechanism described in the"Handling log directory failures" section above – if any partitions become offline and are not correctly assigned to the offline directory in the cluster metadata (from the Broker's point of view), then If the destination directory has failed, it won't be possible to promote the future replica, and the Broker needs to revert the assignment (cancelled locally if still queued). If the source directory has failed, then the future replica might not catch up, and the Controller might not update leadership and ISR for the partition. In this exceptional case, the broker issues a AssignReplicasToDirs RPC to the Controller to both correct the assignment and let it assignment the replica to UUID.OfflineDir - this lets the Controller know that it needs to update leadership and ISR for this partition too.

Controller

Replica placement

For any new partitions, the active controller will use Uuid.UnknownDir as the initial value for log directory UUID for each replicaeach replica – this is the default (empty) value for the tagged field. Each broker with multiple log.dirs hosting replicas then assigns a log directory UUID and communicates it back to the active controller using the new RPC AssignReplicasToDirs so so that cluster metadata can be updated with the log directory assignment. Brokers that are configured with a single log directory to not send this RPC.

Handling log directory failures

...

• Persist a BrokerRegistrationChange record, with the new list of online log directories and update the offline log directories flag.
• Update the Leader and ISR for all the replicas assigned to the failed log directories, persisting PartitionChangeRecords, in a similar way to how leadership and ISR is updated when a broker becomes fenced, unregistered or shuts down.

...

The controller accepts the AssignReplicasToDirs RPC and persists the assignment into metadata records. If the indicated log directory UUID is not a registered log directory then the call fails with error 57 — LOG_DIR_NOT_FOUND .

If the indicated log directory UUID is not one of the Broker's online log directories, then the replica is considered offline and the leader and ISR is updated accordingly, same as when the BrokerHeartbeat indicates a new offline log directory. This should only happen in the exceptional case that a Broker's metadata cache shows an incorrect assignment for some replica during the handling of a failure for the actual directory that hosts that replicadirectory.

Broker registration

Upon a broker registration request the controller will persist the broker registration as cluster metadata including the online log directory list and offline log directories flag for that broker. The controller may receive a new list of online directories and offline log directories flag — different from what was previously persisted in the cluster metadata for the requesting broker.

• If there are no indicated online log directory UUIDs the request is invalid and the controller replies with an error — 42 – INVALID_REQUEST.

• If multiple log directories are registered the broker will remain fenced until the controller learns of all the partition to log directory placements in that broker - i.e. no remaining replicas assigned to Uuid.UnknownDir . The broker will indicate these using the AssignReplicasToDirs RPC.

  • The broker remains fenced by not wanting to unfence itself in heartbeat requests until the number of mismatching replica to log directory assignments is zero. This number is represented by the new metric NumMismatchingReplicaToLogDirAssignments.

  . no remaining replicas assigned to Uuid.UnknownDir . The broker will indicate these using the AssignReplicasToDirs RPC.

  • The broker remains fenced by not wanting to unfence itself in heartbeat requests until the number of mismatching replica to log directory assignments is zero. This number is represented by the new metric NumMismatchingReplicaToLogDirAssignments.
• If multiple log directories are registered and some of them are new (not present in previous registration) then these log directories are assumed to be empty. If they are not, the broker will use the AssignReplicasToDirs RPC to correct assignment and choose not to become UNFENCED before the metadata is correct.
• In the special case where previous broker registration indicates a single online log directory and no offline log directories, and the inbound broker registration request indicates more than one log directory, and one of the indicated log directories is the same one previously registered, then a logical update to all partitions in that broker takes place, assigning the replica's directory to the single directory previously registered – i.e. it is assumed that all replicas are still in the same directory, and this transition to JBOD avoids creating partition change records. This same logic is considered in every node while consuming and caching metadata changesIf multiple log directories are registered and some of them are new (not present in previous registration) then these log directories are assumed to be empty. If they are not, the broker will use the AssignReplicasToDirs  RPC to correct assignment and choose not to become UNFENCED before the metadata is correct.

Brokers whose registration indicates that multiple log directories are configured remain FENCED until all log directory assignments for that broker are learnt by the active controller and persisted into metadata.

...

• Assumed to live in a broker that isn’t yet configured with multiple log directories, and so live in a single log directory, even if the UUID for that directory isn’t yet known by the controller. It is not possible to trigger a log directory failure from a broker that has a single log directory, as the broker would simply shut down if there are no remaining online log directories. Or
• Assigned to a log directory as of yet unknown, in a broker that remains FENCED. As the broker remains FENCED it cannot assume leadership for any partition, and so a log directory failure would be handled by the current partition leader.

...

• Keeping the scope of the log directory to the broker — while this would mean a much simpler change, as was proposed in KIP-589, if only the broker itself knows which partitions were assigned to a log directory, when a log directory fails the broker will need to send a potentially very large request enumerating all the partitions in the failed disk, so that the controller can update leadership and ISRs accordingly. 
• Having the controller determine the log directory for new replicas — this would avoid a further RPC from the broker upon selecting a new log directory for new replicas, and reduce the time until it is safe for the broker to take leadership of the replica. However the broker is in a better position to make a choice of log directory than the controller, as it has easier access to e.g. disk usage in each log directory. The controller could also have this information if the broker were to include it the broker registration. But to keep this change manageable and timely, this optimization is best left for future work. It would also be trickier to manage the ZK→KRaft migration if we had some brokers forwarding the RPC to the controllers and others handling it directly.
• Changing how log directory failure is handled in ZooKeeper mode — ZooKeeper mode is going away, KIP-833 proposed its deprecation in a near future release.
• Using the system path to identify each log directory, or storing the identifier somewhere else — When running Kafka with multiple log directories, each log directory is typically assigned to a different system disk or volume. The same storage device can be made accessible under a different mount, and Kafka should be able to identify the contents as the same disk.  Because the log directory configuration can change on the broker, the only reliable way to identify the log directory in each broker is to add metadata to the file system under the log directory itself.

• Identifying offline log directories. Because we cannot identify them by mount paths, we cannot distinguish between an inaccessible log directory is simply unavailable or if it has been removed from configured – think of a scenario where one log dir is offline and one was removed from log.dirs. In ZK mode we don't care to do this, and we shouldn't do it in KRaft either. What we need to know is if there are any offline log directories, to prevent re-streaming the offline replicas into the remaining online log dirs. In ZK mode, the 'isNew' flag is used to prevent the Broker from creating partitions when any logdir is offline unless they're new. A simple boolean flag to indicate some log dir is offline is enough to maintain the functionality.

...