The Consensus Package provides the Consensus Layer's service definitions and its implementations.
Generally, We maintain multiple copies of application data for the purpose of fault-tolerance and data-integrity. There are variety of consensus algorithms to manage multiple copies of data, which differs in levels of data consistency and performance. Each consensus algorithm may have multiple industrial implementations. The Consensus Layer aims to hide all complications behind different consensus algorithms and implementations, providing higher level of abstraction to the user.
IStateMachineis the user application that manages a local copy of data.Peeris the smallest consensus unit (inside a process), which holds aIStateMachineinternally.ConsensusGroupis a group ofPeerall managing the same copy of data.IConsensusinterface defines the basic functionality provided by Consensus Layer.ConsensusFactoryis the only factory class exposed to other modules to create a consensus layer implementation.
User application can create a ConsensusGroup with k Peer to store data, i.e. that there will be
k copies of data.
When write data into a ConsensusGroup using IConsensus::write, it will be sent to the group
leader's IStateMachine::write . The leader makes decision about this write operation first, then
applies write-operation to the local statemachine using IStateMachine::write, and forward this
operation to other members' IStateMachine::write in the same group
- Define the
IStateMachineto manage local copy of data. - Define the
IConsensusRequestto customize request format - Define the
DataSetto customize the response format - Select a specific "IConsensus" class name and call
ConsensusFactory.getConsensusImpl()to instantiate the corresponding consensus protocol
RatisConsensus is a multi-raft implementation of IConsensus protocol. It is based
on Apache Ratis.
IConsensus consensusImpl =
ConsensusFactory.getConsensusImpl(
ConsensusFactory.RatisConsensus,
new Endpoint(conf.getRpcAddress(), conf.getInternalPort()),
new File(conf.getConsensusDir()),
gid -> new ConfigRegionStateMachine())
.orElseThrow(() ->
new IllegalArgumentException(
String.format(
ConsensusFactory.CONSTRUCT_FAILED_MSG,
ConsensusFactory.RatisConsensus)));
consensusImpl.start();endpointis the communication endpoint for this consensusImpl.StateMachineRegistryIndicates that the consensus layer should generate different state machines for different gidStoreageDirspecifies the location to store RaftLog. Assign a fix location so that the RatisConsensus knows where to recover when crashes and restarts.
ConsensusGroup group = new ConsensusGroup(...);
response = consensusImpl.addConsensusGroup(group.getGroupId(),group.getPeers());The underling consensusImpl will initialize its states, and reaching out to other peers to elect the raft leader.
Notice: this request may fail. It's caller's responsibility to retry / rollback.
suppose now the group contains peer[0,1,2], and we want to remove[1,2] from this group
// the following code should be called in peer both 1 & 2
// first use removePeer to inform the group leader of configuration change
consensusImpl.removePeer(gid,myself);
// then use removeConsensusGroup to clean up local states and data
consensusImpl.removeConsensusGroup(gid);Notice: either of removePeer or removeConsensusGroup may fail. It's caller's responsibility
to retry and make these two calls atomic.
adding a new member is similar to removing a member except that you should call addConsensusGroup
first and then addPeer
// the following code should be called in peer both 1 & 2
// first addConsensusGroup to initialize local states
consensusImpl.addConsensusGroup(gid);
// then use addPeer to inform the previous group members of joining a new member
consensusImpl.addPeer(gid,myself);// pre. For each member newly added, call addConsensusGroup locally to initialize
consensusImpl.changePeer(group.getGroupId(),newGroupmember);
// after. For each member removed, call removeConsensusGroup locally to clean upNotice: the old group and the new group must overlap in at least one member.
ConsensusWriteResponse response = consensusImpl.write(gid,request)
if(response.isSuccess() && response.getStates().code() == 200){
...
}ConsensusReadResponse response = consensusImpl.read(gid,request);
if(response.isSuccess()){
MyDataSet result=(MyDataSet)response.getDataset();
}NOTICE: currently in RatisConsensus, read will direct read the local copy. Thus, the result may be stale and not linearizable!