2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.cluster.raft.behaviors;
11 import akka.actor.ActorRef;
12 import com.google.common.base.Preconditions;
13 import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
14 import org.opendaylight.controller.cluster.raft.RaftActorContext;
15 import org.opendaylight.controller.cluster.raft.RaftReplicator;
16 import org.opendaylight.controller.cluster.raft.RaftState;
17 import org.opendaylight.controller.cluster.raft.internal.messages.SendHeartBeat;
18 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
20 import java.util.Collections;
21 import java.util.HashMap;
22 import java.util.List;
24 import java.util.concurrent.atomic.AtomicLong;
27 * The behavior of a RaftActor when it is in the Leader state
31 * <li> Upon election: send initial empty AppendEntries RPCs
32 * (heartbeat) to each server; repeat during idle periods to
33 * prevent election timeouts (§5.2)
34 * <li> If command received from client: append entry to local log,
35 * respond after entry applied to state machine (§5.3)
36 * <li> If last log index ≥ nextIndex for a follower: send
37 * AppendEntries RPC with log entries starting at nextIndex
39 * <li> If successful: update nextIndex and matchIndex for
41 * <li> If AppendEntries fails because of log inconsistency:
42 * decrement nextIndex and retry (§5.3)
44 * <li> If there exists an N such that N > commitIndex, a majority
45 * of matchIndex[i] ≥ N, and log[N].term == currentTerm:
46 * set commitIndex = N (§5.3, §5.4).
48 public class Leader extends AbstractRaftActorBehavior {
51 private final Map<String, ActorRef> followerToReplicator = new HashMap<>();
53 public Leader(RaftActorContext context, List<String> followers){
56 for(String follower : followers) {
58 ActorRef replicator = context.actorOf(
60 new FollowerLogInformationImpl(follower,
67 // Create a replicator for each follower
68 followerToReplicator.put(follower, replicator);
74 @Override public RaftState handleMessage(ActorRef sender, Object message) {
75 Preconditions.checkNotNull(sender, "sender should not be null");
77 if(message instanceof SendHeartBeat) {
78 sender.tell(new AppendEntries(
79 context.getTermInformation().getCurrentTerm().get() , context.getId(),
80 context.getReplicatedLog().last().getIndex(),
81 context.getReplicatedLog().last().getTerm(),
82 Collections.EMPTY_LIST), context.getActor());
84 return RaftState.Leader;