Distributed consensus with jgroups-raft

The architecture of jgroups-raft is shown below.

The components that make up jgroups-raft are

The figure above shows one node in a cluster, but the other nodes have the same setup except that every node is required to have a different raft_id (defined in RAFT). This is a string which defines one cluster member; all members need to have different raft_ids (more on this later).

Each cluster member has an address (a UUID) assigned by JGroups and a raft_id which needs to be assigned by the user. The latter is a string (e.g. "A") which needs to be unique in the entire cluster. In other words, the raft_id is the identity of a member for the sake of jgroups-raft.

A Raft cluster has a fixed size, so that a majority can be computed for leader election and appending of entries. The members allowed into the cluster is defined in RAFT.members, e.g.

This defines a cluster of 3 members: "A", "B" and "C" (whose majority is 2).

These are the raft_id attributes of the 3 members, so attribute raft_id in the example above needs to be one of them. If we don’t start this member with the system property -Draft_id=X (where X needs to be "A", "B", or "C"), then the member will start up as "undefined" is not a member of {"A", "B", "C"}.

Attribute raft_id is also used to define the location of the persistent log; unless log_name is defined in RAFT, the location is computed as <temp_dir>/<raft_id>.log, e.g. /tmp/A.log.

Note that members can be added and removed dynamically (without taking the entire cluster down, changing the configuration and restarting it), see Adding and removing members dynamically.

As shown in The architecture of jgroups-raft, RaftHandle is the main class users will be dealing with. It provides methods to change state (append entries) in the replicated state machines, and a state machine can be registered with it. The state machine will be initialized at startup and updated by jgroups-raft whenever there is a change.

A successful change is committed to the persistent logs of all cluster members and applied to their state machines, so all state machines have exactly the same state.

The configuration of a member is either done declaratively via an XML config file or programmatically. Refer to the JGroups documentation for details.

A sample XML configuration file is shown below (edited for brevity):

This is a regular JGroups XML configuration, except that jgroups-raft added a few additional protocols.

The RAFT protocol provides methods addServer(String raft_id) and removeServer(String raft_id) to add and remove servers from the static membership (defined by RAFT.members). Only one server at a time can be added and removed, and adding or removing a server needs a majority ack to be committed.

Both methods are exposed via JMX, so jconsole could be used. However, jgroups-raft also provides a script (client.sh) to do this in a more convenient way. The script uses Client to connect to a member’s CLIENT protocol running at localhost:1965 (can be changed). The request is then forwarded to the current leader.

The steps to add a member are as follows (say we have RAFT.members="A,B,C" and want to add "D"):

ReplicatedStateMachine is a key-value store replicating its contents to all cluster members. Contrary to the JGroups equivalent (ReplicatedHashMap), changes are replicated by consensus and logged to a persistent log.

While the JGroups version is allowed to make progress during network partitions, and users need to merge possibly diverging state from different partitions after a partition heals, ReplicatedStateMachine will allow progress only in the majority partition, so no state merging needs to be done after a partition heals.

Not having to merge state is certainly simpler, but comes at the expense of availability: if N/2+1 members leave or split into different partitions, ReplicatedStateMachine will be unavailable (all requests will time out).

However, the advantage is that the members' states will never diverge.

ReplicatedStateMachine requires a JChannel in its constructor and has put(), get() and remove() methods. The code below shows how to create an instance of ReplicatedStateMachine and add an element to it:

There’s a demo ReplicatedStateMachineDemo which can be used to interactively use ReplicatedStateMachine.

CounterService provides a replicated counter which can get be set, get and compare-and-set, implementing JGroups' Counter interface:

A Counter implementation is created via the CounterService building block (edited):

CounterService is mainly used to get an existing or create a new Counter implementation (getOrCreateCounter()), or to delete an existing counter (deleteCounter()).

To create an instance of CounterService, a JChannel has to be passed to the constructor. The sample code below shows how to use this:

Any member in the cluster can change the same counter and all operations are ordered by the Raft leader, which causes the replicated counters to have exactly the same value in all members.

Comparing this to the JGroups equivalent, a jgroups-raft counter never diverges in different members, again at the expense of availability. In the JGroups version, counters are always available, but may diverge, e.g. in a split brain scenario, and have to be reconciled by the application after the split brain is resolved.

There’s a demo CounterServiceDemo which can be used to interactively manipulate replicated counters.

A singleton service is a service which is supposed to run only once in an entire cluster. Typically, in JGroups, a singleton service is started on the first member of a cluster. For example, if we have {A,B,C,D,E}, the singleton service (or services) would be running on A.

If we have a partition, such that the cluster falls apart into {A,B,C} and {D,E}, then an additional singleton would be started on D, as D became coordinator and doesn’t know {A,B,C} didn’t leave, but were partitioned away instead.

When the partition ends, if D is not coordinator anymore, it would stop its singleton services.

If multiple singletons (as provided by JGroups, e.g. during a network split) cannot be tolerated by the application, and the application has a requirement that at most one singleton service can be running (better none than two), jgroups-raft can be used.

The mechanism to implement singleton services in jgroups-raft is leader election: it is guaranteed that at most one leader exists in a given cluster at the same time. This is exactly what we need for singletons. The code below shows how to do this:

This protocol prevents duplicate members from joining the cluster. The protocol needs to be located somewhere below GMS.

NO_DUPES catches JOIN requests from a joiner to the JGroups coordinator and checks if the joiner’s raft_id is already contained in the current membership, and rejects the JOIN if this is the case.

For example, if we have current members {A,B} and another member with raft_id "B" joins, then the joiner would get the following exception when trying to join the cluster:

The error message is SecurityException: join of B rejected as it would create a view with duplicate members (current view: [B|1] (2) [B, A]), which shows that view {B,A} already contains a member with raft_id B, and so the JOIN request of the new member is rejected.

ELECTION is the protocol which performs leader election, as defined by Raft. Its attributes define the election timeout and the heartbeat interval (see Raft for details).

RAFT is the main protocol in jgroups-raft; it implements log appending and committing, snapshotting and log compaction, syncing of new members and so on.

The REDIRECT protocol needs to be somewhere above RAFT. It keeps track of the current Raft leader and redirects requests to the right leader. If there is no leader, e.g. because there’s no majority to elect one, an exception will be thrown.

CLIENT listens on a socket for client requests. When a request is received, it is sent down where it will be forwarded (by REDIRECT) to the current leader which executes the request. The responses is then sent back to the client.