Kudu has many RPCs that need error handling. When an RPC fails due to a timeout or network failure, the sender needs to retry it in the same or in a different server, but it often doesn't know whether the original RPC succeeded. Certain RPCs are not idempotent i.e. if handled twice they will lead to incorrect state and thus we need a mechanism by which the sender can retry the RPC but the receiver is guaranteed to execute it Exactly Once.
[1] introduces 4 subproblems that need to be solved to obtain exactly-once semantics on replicated RPCs, we'll use the same terminology to avoid redefining the concepts:
We can address these problems in multiple ways and, in particular, at multiple abstraction layers:
Completely encapsulated at the RPC layer - This is the option that [1] presents at length, allowing to “bolt-on” these properties almost independently of what the underlying RPCs are. This option, while very general, would require a lot of work particularly as for replicated RPCs we‘d need to come up with a generic, durable, Completion Record storage mechanism. While “elegant” in some ways this option also seems weird in other ways: for instance it’s not totally clear what happens when an RPC mutates multiple “server objects” or what these server objects really are.
Handled ad-hoc by the client and the tablet server/master, outside of the RPC layer - With this option, we choose to handle errors at the TabletServer/TabletReplica, ad-hoc, and for each different RPC. For specific RPCs like Write() this option seems to map to existing components quite well. For instance the completion record can be the raft log and replica replay would mean that it would be relatively easy to implement a retry rendezvous mechanism. However this option lacks in generality and it would likely require duplicate error handling for operations that are not tablet server transactions.
A hybrid of the above - Certain parts of the problem would be handled by the RPC layer, such as retrying and retry rendez-vous logic, but other parts of the logic would be implemented ad-hoc in other layers. For instance, for Write()s, the RPC layer would know how to sequence and retry the RPCs but would delegate durability and cross-server replicated operation rendezvous to the tablet replica/transaction manager.
We chose option 3, in which most of the sub-component is encapsulated in the RPC layer, allowing to provide limited (non-persistent) bolt-on exactly once semantics for generic RPCs, but the API is exposed so that we can perform ad-hoc handling of persistence where required (e.g. for Write RPCs).
We addressed each of the four sub-problems mentioned in the previous section the following way:
Each individual RPC attempt contains the following information:
CompletionRecord - A completion record of an RPC is the response that is sent back to the client once the RPC is complete. Once a response for a request is built, it is cached in the RPC subsystem and clients will always receive the same response for the same request. The RPC subsystem does not handle response durability though, it is up to the ad-hoc integration to make sure that either responses are durably stored or can be rebuilt, consistently.
Retry rendezvous - We make sure that multiple attempts at the same RPC in the same server meet by making them go through a new component, the ResultTracker. This component is responsible for figuring out the state of the RPC, among:
We opted for not implementing the distributed client lease mechanism in [1]. This does allow for cases where an RPC is executed twice: for instance if a client attempts an RPC and then is silent for a period greater than the time-based garbage collection period before re-attempting the same RPC. We consider that this is unlikely enough not to justify implementing a complex lease management mechanism, but for documentation purposes the following scenario could cause double execution:
Clients assign themselves a UUID that acts as their unique identifier for their lifetime. Clients have a component, the Request Tracker that is responsible for assigning new RPC Sequence numbers and tracking which ones are outstanding. Along with the outstanding requests, the Request Tracker is responsible for tracking the First incomplete sequence number, i.e. the id of the first outstanding RPC. This is important for garbage collection purposes, as we'll see later. The client system is also responsible for keeping a counter of the times an RPC is attempted and making sure that each time an attempt is performed it is assigned a new AttemptNumber based on this counter. Together these four elements form a Request Id and are set in the RPC header that takes the form:
========================================= | RequestIdPB | ---------------------------------------- | - client_id : string | | - seq_no : int64 | | - first_incomplete_seq_no : int64 | | - attempt_no : int64 | =========================================
When an RPC reaches the server, the Request Id is passed to a ResultTracker the server-side component responsible for storing Completion Records (the response protobufs), and for doing the Retry Rendezvous. For each of the RPCs this component will determine which of the following states it is in:
If the ResultTracker returns that the request is NEW, then execution proceeds, for instance in the case of a Write() this means that it will be sent to the TransactionManager for replication and eventually to be applied to the tablet.
Once execution of the request is completed successfully it is sent to the ResultTracker which will store the response in memory (to be able to reply to future retries) and reply back to the client.
The response for errors is not stored. This is for two reasons: Errors (are not supposed) to have side-effects; Errors might be transient, for instance a write to a non-leader replica may fail but a retry might be successful if the replica is elected leader.
The mechanism above, just by itself, does not handle replicated RPCs. If an RPC is replicated, i.e. if an RPC must be executed exactly once across a set of nodes (e.g. Write()), its idempotency is not totally covered by the mechanism above.
Responses are not stored persistently so, for replicated RPCs, implementations must take care that the same request always has the same response, i.e. that the exact same response can be rebuilt from history.
To make sure the results of an RPC are tracked transiently on the server side (we'll cover how we made sure that write results were tracked persistently in the following section), all that is required is that the service definition enables the appropriate option. For instance in the case of writes this is done the following way:
service TabletServerService {
...
rpc Write(WriteRequestPB) returns (WriteResponsePB) {
option (kudu.rpc.track_rpc_result) = true;
}
...
}
RPCs that are replicated, for fault tolerance, require more than the mechanics above. In particular they have the following additional requirements:
Cross-replica rendezvous - The ResultTracker makes sure that, for a single server, all attempts of an RPC from a client directly to that server serialized. However this does not take into account attempts from the same client to other servers, which must also be serialized in some way.
Response rebuilding - When a server crashes it loses some in-memory state. Requests for completed operations are durably stored on-disk so that that volatile state can be rebuilt, but responses are not. When a request is replayed to rebuild lost state, it's response must be stored again and it must be the same as the original response sent to the client.
Writes (any operation that mutates rows on a tablet server) are the primary use case for exactly once semantics and thus it was implemented first (for reference this landed in commit 6d2679bd and this is the corresponding gerrit). Examining this change is likely to enlighten adding Exactly Once semantics to other replicated RPCs. A lot of the changes were mechanical, the relevant ones are the following:
The cross-replica retry rendezvous is implemented in transaction_driver.{h,cc}, its inner workings are detailed in the header. This basically makes sure that when an RPC is received from a client and a different attempt of the same RPC is received from another replica (a previous leader), we execute only one of those attempts (the replica one).
Response rebuilding is implemented in tablet_bootstrap.cc, which basically adds the ability to rebuild the original response, when a request is replayed on tablet bootstrap.
[1]Implementing Linearizability at Large Scale and Low Latency