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.
IStateMachine
is the user application that manages a local copy of data.Peer
is the smallest consensus unit (inside a process), which holds a IStateMachine
internally.ConsensusGroup
is a group of Peer
all managing the same copy of data.IConsensus
interface defines the basic functionality provided by Consensus Layer.ConsensusFactory
is 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
IStateMachine
to manage local copy of data.IConsensusRequest
to customize request formatDataSet
to customize the response formatConsensusFactory.getConsensusImpl()
to instantiate the corresponding consensus protocolRatisConsensus 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();
endpoint
is the communication endpoint for this consensusImpl.StateMachineRegistry
Indicates that the consensus layer should generate different state machines for different gidStoreageDir
specifies 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 up
Notice: 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!