iotdb-core/consensus/README.md
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!