@Test public void testBecomePreLeaderOnReceivingMajorityVotesInThreeNodeCluster() { MockRaftActorContext raftActorContext = createActorContext(); raftActorContext.setLastApplied(-1); raftActorContext.setPeerAddresses(setupPeers(2)); candidate = new Candidate(raftActorContext); candidate = candidate.handleMessage(peerActors[0], new RequestVoteReply(1, true)); assertEquals("Behavior", RaftState.PreLeader, candidate.state()); }
@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();  }