@Test public void appendEntries() throws java.lang.Exception { for (Server s : raftServer) { s.start(); } raftServer[0].become(CANDIDATE, FOLLOWER); assertTrue(raftServer[verifyLeaderElected(raftServer)] == raftServer[0]); String methodName = "public java.lang.String java.lang.Object.toString()"; ArrayList<Object> args = new ArrayList<Object>(); Object obj = new ConsensusRSMPolicy.RPC(methodName, args); assertEquals(0, raftServer[0].pState.log().size()); assertEquals(0, raftServer[1].pState.log().size()); assertEquals(0, raftServer[2].pState.log().size()); raftServer[0].applyToStateMachine(obj); verifyLastApplied(raftServer[0], raftServer[1]); verifyLastApplied(raftServer[0], raftServer[2]); raftServer[0].applyToStateMachine(obj); verifyLastApplied(raftServer[0], raftServer[1]); verifyLastApplied(raftServer[0], raftServer[2]); for (int i = 0; i < raftServer[0].pState.log().size(); i++) { LogEntry leaderEntry = raftServer[0].pState.log().get(i); LogEntry entry1 = raftServer[1].pState.log().get(i); LogEntry entry2 = raftServer[2].pState.log().get(i); assertEquals(leaderEntry.operation, entry1.operation); assertEquals(leaderEntry.term, entry1.term); assertEquals(leaderEntry.operation, entry2.operation); assertEquals(leaderEntry.term, entry2.term); } assertEquals(raftServer[0].vState.getLastApplied(), raftServer[1].vState.getLastApplied()); assertEquals(raftServer[0].vState.getLastApplied(), raftServer[2].vState.getLastApplied()); }
public int appendEntries( int term, UUID leader, int prevLogIndex, int prevLogTerm, List<LogEntry> entries, int leaderCommit) throws InvalidTermException, PrevLogTermMismatch, InvalidLogIndex { logger.fine( String.format( "%s: received AppendEntries request from leader %s, term %d, prevLogIndex=%d, prevLogTerm=%d, leaderCommit=%d, entries=%d", pState.myServerID, leader, term, prevLogIndex, prevLogTerm, leaderCommit, entries.size())); respondToRemoteTerm(term); if (term < pState.getCurrentTerm()) { throw new InvalidTermException( "Server: Attempt to append entries from prior leader term " + term + ", current term " + pState.getCurrentTerm(), pState.getCurrentTerm()); } vState.setCurrentLeader(leader); synchronized (leaderHeartbeatReceiveTimer) { leaderHeartbeatReceiveTimer.reset(); } synchronized (pState) { LogEntry prevLogEntry; if (prevLogIndex >= 0) { try { prevLogEntry = pState.log().get(prevLogIndex); } catch (IndexOutOfBoundsException e) { throw new InvalidLogIndex( "Attempt to append entry with invalid previous log index: " + prevLogIndex, prevLogIndex); } if (prevLogEntry.term != prevLogTerm) { throw new PrevLogTermMismatch( "Attempt to append entry with invalid previous log term. Requested term " + prevLogTerm + ", actual term: " + prevLogEntry.term, prevLogIndex, prevLogEntry.term, prevLogTerm); } } int logIndex = prevLogIndex; if (pState.log().size() - 1 >= ++logIndex) { logger.fine( String.format( "%s: Removing conflicting log entries. Current log size=%d, Current commit index=%d. Replacing logs starting from index=%d", pState.myServerID, pState.log().size(), this.vState.getCommitIndex(), logIndex)); pState.setLog(pState.log().subList(0, logIndex)); } pState.log().addAll(entries); if (leaderCommit > vState.getCommitIndex()) { vState.setCommitIndex( Math.min(leaderCommit, pState.log().size() - 1), vState.getCommitIndex()); } } applyCommitted(); return pState.getCurrentTerm(); }
Server implements RemoteRaftServer { public int appendEntries( int term, UUID leader, int prevLogIndex, int prevLogTerm, List<LogEntry> entries, int leaderCommit) throws InvalidTermException, PrevLogTermMismatch, InvalidLogIndex { logger.fine( String.format( "%s: received AppendEntries request from leader %s, term %d, prevLogIndex=%d, prevLogTerm=%d, leaderCommit=%d, entries=%d", pState.myServerID, leader, term, prevLogIndex, prevLogTerm, leaderCommit, entries.size())); respondToRemoteTerm(term); if (term < pState.getCurrentTerm()) { throw new InvalidTermException( "Server: Attempt to append entries from prior leader term " + term + ", current term " + pState.getCurrentTerm(), pState.getCurrentTerm()); } vState.setCurrentLeader(leader); synchronized (leaderHeartbeatReceiveTimer) { leaderHeartbeatReceiveTimer.reset(); } synchronized (pState) { LogEntry prevLogEntry; if (prevLogIndex >= 0) { try { prevLogEntry = pState.log().get(prevLogIndex); } catch (IndexOutOfBoundsException e) { throw new InvalidLogIndex( "Attempt to append entry with invalid previous log index: " + prevLogIndex, prevLogIndex); } if (prevLogEntry.term != prevLogTerm) { throw new PrevLogTermMismatch( "Attempt to append entry with invalid previous log term. Requested term " + prevLogTerm + ", actual term: " + prevLogEntry.term, prevLogIndex, prevLogEntry.term, prevLogTerm); } } int logIndex = prevLogIndex; if (pState.log().size() - 1 >= ++logIndex) { logger.fine( String.format( "%s: Removing conflicting log entries. Current log size=%d, Current commit index=%d. Replacing logs starting from index=%d", pState.myServerID, pState.log().size(), this.vState.getCommitIndex(), logIndex)); pState.setLog(pState.log().subList(0, logIndex)); } pState.log().addAll(entries); if (leaderCommit > vState.getCommitIndex()) { vState.setCommitIndex( Math.min(leaderCommit, pState.log().size() - 1), vState.getCommitIndex()); } } applyCommitted(); return pState.getCurrentTerm(); } }
Server implements RemoteRaftServer { public int appendEntries( int term, UUID leader, int prevLogIndex, int prevLogTerm, List<LogEntry> entries, int leaderCommit) throws InvalidTermException, PrevLogTermMismatch, InvalidLogIndex { logger.fine( String.format( "%s: received AppendEntries request from leader %s, term %d, prevLogIndex=%d, prevLogTerm=%d, leaderCommit=%d, entries=%d", pState.myServerID, leader, term, prevLogIndex, prevLogTerm, leaderCommit, entries.size())); respondToRemoteTerm(term); if (term < pState.getCurrentTerm()) { throw new InvalidTermException( "Server: Attempt to append entries from prior leader term " + term + ", current term " + pState.getCurrentTerm(), pState.getCurrentTerm()); } vState.setCurrentLeader(leader); synchronized (leaderHeartbeatReceiveTimer) { leaderHeartbeatReceiveTimer.reset(); } synchronized (pState) { LogEntry prevLogEntry; if (prevLogIndex >= 0) { try { prevLogEntry = pState.log().get(prevLogIndex); } catch (IndexOutOfBoundsException e) { throw new InvalidLogIndex( "Attempt to append entry with invalid previous log index: " + prevLogIndex, prevLogIndex); } if (prevLogEntry.term != prevLogTerm) { throw new PrevLogTermMismatch( "Attempt to append entry with invalid previous log term. Requested term " + prevLogTerm + ", actual term: " + prevLogEntry.term, prevLogIndex, prevLogEntry.term, prevLogTerm); } } int logIndex = prevLogIndex; if (pState.log().size() - 1 >= ++logIndex) { logger.fine( String.format( "%s: Removing conflicting log entries. Current log size=%d, Current commit index=%d. Replacing logs starting from index=%d", pState.myServerID, pState.log().size(), this.vState.getCommitIndex(), logIndex)); pState.setLog(pState.log().subList(0, logIndex)); } pState.log().addAll(entries); if (leaderCommit > vState.getCommitIndex()) { vState.setCommitIndex( Math.min(leaderCommit, pState.log().size() - 1), vState.getCommitIndex()); } } applyCommitted(); return pState.getCurrentTerm(); } Server(StateMachineApplier applier); }
Server implements RemoteRaftServer { public int appendEntries( int term, UUID leader, int prevLogIndex, int prevLogTerm, List<LogEntry> entries, int leaderCommit) throws InvalidTermException, PrevLogTermMismatch, InvalidLogIndex { logger.fine( String.format( "%s: received AppendEntries request from leader %s, term %d, prevLogIndex=%d, prevLogTerm=%d, leaderCommit=%d, entries=%d", pState.myServerID, leader, term, prevLogIndex, prevLogTerm, leaderCommit, entries.size())); respondToRemoteTerm(term); if (term < pState.getCurrentTerm()) { throw new InvalidTermException( "Server: Attempt to append entries from prior leader term " + term + ", current term " + pState.getCurrentTerm(), pState.getCurrentTerm()); } vState.setCurrentLeader(leader); synchronized (leaderHeartbeatReceiveTimer) { leaderHeartbeatReceiveTimer.reset(); } synchronized (pState) { LogEntry prevLogEntry; if (prevLogIndex >= 0) { try { prevLogEntry = pState.log().get(prevLogIndex); } catch (IndexOutOfBoundsException e) { throw new InvalidLogIndex( "Attempt to append entry with invalid previous log index: " + prevLogIndex, prevLogIndex); } if (prevLogEntry.term != prevLogTerm) { throw new PrevLogTermMismatch( "Attempt to append entry with invalid previous log term. Requested term " + prevLogTerm + ", actual term: " + prevLogEntry.term, prevLogIndex, prevLogEntry.term, prevLogTerm); } } int logIndex = prevLogIndex; if (pState.log().size() - 1 >= ++logIndex) { logger.fine( String.format( "%s: Removing conflicting log entries. Current log size=%d, Current commit index=%d. Replacing logs starting from index=%d", pState.myServerID, pState.log().size(), this.vState.getCommitIndex(), logIndex)); pState.setLog(pState.log().subList(0, logIndex)); } pState.log().addAll(entries); if (leaderCommit > vState.getCommitIndex()) { vState.setCommitIndex( Math.min(leaderCommit, pState.log().size() - 1), vState.getCommitIndex()); } } applyCommitted(); return pState.getCurrentTerm(); } Server(StateMachineApplier applier); UUID getMyServerID(); void start(); void stop(); int appendEntries(
            int term,
            UUID leader,
            int prevLogIndex,
            int prevLogTerm,
            List<LogEntry> entries,
            int leaderCommit); int requestVote(int term, UUID candidate, int lastLogIndex, int lastLogTerm); Object applyToStateMachine(Object operation); void addServer(UUID id, RemoteRaftServer server); }
Server implements RemoteRaftServer { public int appendEntries( int term, UUID leader, int prevLogIndex, int prevLogTerm, List<LogEntry> entries, int leaderCommit) throws InvalidTermException, PrevLogTermMismatch, InvalidLogIndex { logger.fine( String.format( "%s: received AppendEntries request from leader %s, term %d, prevLogIndex=%d, prevLogTerm=%d, leaderCommit=%d, entries=%d", pState.myServerID, leader, term, prevLogIndex, prevLogTerm, leaderCommit, entries.size())); respondToRemoteTerm(term); if (term < pState.getCurrentTerm()) { throw new InvalidTermException( "Server: Attempt to append entries from prior leader term " + term + ", current term " + pState.getCurrentTerm(), pState.getCurrentTerm()); } vState.setCurrentLeader(leader); synchronized (leaderHeartbeatReceiveTimer) { leaderHeartbeatReceiveTimer.reset(); } synchronized (pState) { LogEntry prevLogEntry; if (prevLogIndex >= 0) { try { prevLogEntry = pState.log().get(prevLogIndex); } catch (IndexOutOfBoundsException e) { throw new InvalidLogIndex( "Attempt to append entry with invalid previous log index: " + prevLogIndex, prevLogIndex); } if (prevLogEntry.term != prevLogTerm) { throw new PrevLogTermMismatch( "Attempt to append entry with invalid previous log term. Requested term " + prevLogTerm + ", actual term: " + prevLogEntry.term, prevLogIndex, prevLogEntry.term, prevLogTerm); } } int logIndex = prevLogIndex; if (pState.log().size() - 1 >= ++logIndex) { logger.fine( String.format( "%s: Removing conflicting log entries. Current log size=%d, Current commit index=%d. Replacing logs starting from index=%d", pState.myServerID, pState.log().size(), this.vState.getCommitIndex(), logIndex)); pState.setLog(pState.log().subList(0, logIndex)); } pState.log().addAll(entries); if (leaderCommit > vState.getCommitIndex()) { vState.setCommitIndex( Math.min(leaderCommit, pState.log().size() - 1), vState.getCommitIndex()); } } applyCommitted(); return pState.getCurrentTerm(); } Server(StateMachineApplier applier); UUID getMyServerID(); void start(); void stop(); int appendEntries(
            int term,
            UUID leader,
            int prevLogIndex,
            int prevLogTerm,
            List<LogEntry> entries,
            int leaderCommit); int requestVote(int term, UUID candidate, int lastLogIndex, int lastLogTerm); Object applyToStateMachine(Object operation); void addServer(UUID id, RemoteRaftServer server); static final int LEADER_HEARTBEAT_TIMEOUT; }