@Test @Override public void testHandleAppendEntriesAddSameEntryToLog() { MockRaftActorContext context = createActorContext(); context.getTermInformation().update(2, "test"); MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("zero"); setLastLogEntry(context, 2, 0, payload); List<ReplicatedLogEntry> entries = new ArrayList<>(); entries.add(new SimpleReplicatedLogEntry(0, 2, payload)); final AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 2, -1, (short)0); behavior = createBehavior(context); context.getTermInformation().update(2, "test"); behavior.handleMessage(candidateActor, "unknown"); RaftActorBehavior raftBehavior = behavior.handleMessage(candidateActor, appendEntries); assertEquals("Raft state", RaftState.Follower, raftBehavior.state()); assertEquals("ReplicatedLog size", 1, context.getReplicatedLog().size()); handleAppendEntriesAddSameEntryToLogReply(candidateActor); }
@Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout) { log.debug("{}: Received ElectionTimeout", logName()); if (votesRequired == 0) { return internalSwitchBehavior(RaftState.Leader); } startNewTerm(); scheduleElection(electionDuration()); return this; } if (message instanceof RaftRPC) { RaftRPC rpc = (RaftRPC) message; log.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc, context.getTermInformation().getCurrentTerm()); if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) { log.info("{}: Term {} in \"{}\" message is greater than Candidate's term {} - switching to Follower", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); if (message instanceof RequestVote) { super.handleMessage(sender, message); } return internalSwitchBehavior(RaftState.Follower); } } return super.handleMessage(sender, message); }
Candidate extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout) { log.debug("{}: Received ElectionTimeout", logName()); if (votesRequired == 0) { return internalSwitchBehavior(RaftState.Leader); } startNewTerm(); scheduleElection(electionDuration()); return this; } if (message instanceof RaftRPC) { RaftRPC rpc = (RaftRPC) message; log.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc, context.getTermInformation().getCurrentTerm()); if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) { log.info("{}: Term {} in \"{}\" message is greater than Candidate's term {} - switching to Follower", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); if (message instanceof RequestVote) { super.handleMessage(sender, message); } return internalSwitchBehavior(RaftState.Follower); } } return super.handleMessage(sender, message); } }
Candidate extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout) { log.debug("{}: Received ElectionTimeout", logName()); if (votesRequired == 0) { return internalSwitchBehavior(RaftState.Leader); } startNewTerm(); scheduleElection(electionDuration()); return this; } if (message instanceof RaftRPC) { RaftRPC rpc = (RaftRPC) message; log.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc, context.getTermInformation().getCurrentTerm()); if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) { log.info("{}: Term {} in \"{}\" message is greater than Candidate's term {} - switching to Follower", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); if (message instanceof RequestVote) { super.handleMessage(sender, message); } return internalSwitchBehavior(RaftState.Follower); } } return super.handleMessage(sender, message); } Candidate(final RaftActorContext context); }
Candidate extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout) { log.debug("{}: Received ElectionTimeout", logName()); if (votesRequired == 0) { return internalSwitchBehavior(RaftState.Leader); } startNewTerm(); scheduleElection(electionDuration()); return this; } if (message instanceof RaftRPC) { RaftRPC rpc = (RaftRPC) message; log.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc, context.getTermInformation().getCurrentTerm()); if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) { log.info("{}: Term {} in \"{}\" message is greater than Candidate's term {} - switching to Follower", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); if (message instanceof RequestVote) { super.handleMessage(sender, message); } return internalSwitchBehavior(RaftState.Follower); } } return super.handleMessage(sender, message); } Candidate(final RaftActorContext context); @Override final String getLeaderId(); @Override final short getLeaderPayloadVersion(); @Override RaftActorBehavior handleMessage(final ActorRef sender, final Object message); @Override void close(); }
Candidate extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout) { log.debug("{}: Received ElectionTimeout", logName()); if (votesRequired == 0) { return internalSwitchBehavior(RaftState.Leader); } startNewTerm(); scheduleElection(electionDuration()); return this; } if (message instanceof RaftRPC) { RaftRPC rpc = (RaftRPC) message; log.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc, context.getTermInformation().getCurrentTerm()); if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) { log.info("{}: Term {} in \"{}\" message is greater than Candidate's term {} - switching to Follower", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); if (message instanceof RequestVote) { super.handleMessage(sender, message); } return internalSwitchBehavior(RaftState.Follower); } } return super.handleMessage(sender, message); } Candidate(final RaftActorContext context); @Override final String getLeaderId(); @Override final short getLeaderPayloadVersion(); @Override RaftActorBehavior handleMessage(final ActorRef sender, final Object message); @Override void close();  }