...

For the KIP-500 bridge release (version 2.6.0 as of the time of this proposal), brokers will be allowed to read from ZooKeeper, but only the controller will be allowed to write. Since we will not be able to use ZooKeeper as an event bus between brokers, we must come up with an alternate strategy.

With this KIP, we propose to add a new RPC that allows a broker to directly communicate state changes of a replica to the controller. This will replace the ZooKeeper based notification for log dir failures and potentially could replace the existing controlled shutdown RPC. Since this RPC is somewhat generic, it could also be used to mark a replicas a "online" following some kind of log dir recovery procedure (out of scope for this proposal).

Public Interfaces

We will add a new RPC named AlterReplicaState which requires CLUSTER_ACTION permissions

...

Upon encountering errors relating to the log dirs, the broker will now send an RPC to the controller indicating that one or more replicas need to be marked offline. Since these types of errors are likely to occur for group of replicas at once, we will continue to use a background thread in ReplicaManager to allow these errors to accumulate before sending a message to the controller. The controller will synchronously perform basic validation of the request (permissions, do check if topics existsexist, etc) and asynchronously perform the necessary actions to process the replica state changes. 

Previously, the broker would only send its ID to the controller using the ZK watch mechanism. Since only the broker ID was sent (and not the broker epoch) the controller needed to probe the broker to learn its true state. Theoretically, a broker could have been repaired and restarted before the controller had a chance to react to the event. The controller would probe the brokers using LeaderAndIsr requests to learn in what state the replicas were in. In this proposal, we now include the broker ID and epoch in the request, so the controller can safely update its internal replica state based on the request data. If the broker had, in fact, been restarted since sending the AlterReplicaState, the controller would be gated by the broker epoch and would not take any action. 

...

• The NewState field in the request is will only support the value 0x1 which will represent "offline"
• The Reason field is an optional a textual description of why the event is being sentIf no Topic is given, it is implied that all topics on this broker are being indicated
• If a Topic and no partitions are given, it is implied that all partitions of this topic are being indicated

Failure Modes

Like the ZK watch before, the AlterReplicaState RPC from the broker to the controller is best-effort. We are not guaranteed to be able to send a message to the controller to indicate a replica state change. Also, since the processing of this RPC by the controller is asynchronous, we are not guaranteed that the subsequent handling of the state change will happen. The following failure scenarios are possible:

...

Another rejected approach was to add an RPC which mirrored the JSON payload used by the ZK workflow currently implemented. This was rejected in favor of a more generic RPC that could be used for other purposes in the future. It was also rejected to prevent "leaking" the notion of a log dir to the public API and to the Controller.

Future Work

We have intentionally designed this KIP to accommodate future use cases involving the replica state. For example, we would potentially mark a replicas a "online" following some kind of log dir recovery procedure.

This RPC is quite similar to the existing ControlledShutdown RPC. Since If we intend to interpret an empty extend this RPC to define a null list of topics to mean all topics, the proposed AlterReplicaState RPC we could subsume the ControlledShutdown RPC with the AlterReplicaState RPC. After we implement this new RPC, we can consider if we want to move forward and consolidate ControlledShutdown into AlterReplicaState as a future KIP.

...