BUG-7033: Add batchHint flag to RaftActor#persistData
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / RaftActorServerConfigurationSupport.java
1 /*
2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.controller.cluster.raft;
9
10 import akka.actor.ActorRef;
11 import akka.actor.ActorSelection;
12 import akka.actor.Cancellable;
13 import com.google.common.base.Preconditions;
14 import java.util.ArrayDeque;
15 import java.util.ArrayList;
16 import java.util.Collection;
17 import java.util.HashSet;
18 import java.util.List;
19 import java.util.Map;
20 import java.util.Queue;
21 import java.util.UUID;
22 import javax.annotation.Nullable;
23 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
24 import org.opendaylight.controller.cluster.raft.base.messages.SnapshotComplete;
25 import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
26 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader;
27 import org.opendaylight.controller.cluster.raft.messages.AddServer;
28 import org.opendaylight.controller.cluster.raft.messages.AddServerReply;
29 import org.opendaylight.controller.cluster.raft.messages.ChangeServersVotingStatus;
30 import org.opendaylight.controller.cluster.raft.messages.RemoveServer;
31 import org.opendaylight.controller.cluster.raft.messages.RemoveServerReply;
32 import org.opendaylight.controller.cluster.raft.messages.ServerChangeReply;
33 import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
34 import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
35 import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSnapshotReply;
36 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
37 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
38 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
39 import org.opendaylight.yangtools.concepts.Identifier;
40 import org.opendaylight.yangtools.util.AbstractUUIDIdentifier;
41 import org.slf4j.Logger;
42 import org.slf4j.LoggerFactory;
43 import scala.concurrent.duration.FiniteDuration;
44
45 /**
46  * Handles server configuration related messages for a RaftActor.
47  *
48  * @author Thomas Pantelis
49  */
50 class RaftActorServerConfigurationSupport {
51     private static final Logger LOG = LoggerFactory.getLogger(RaftActorServerConfigurationSupport.class);
52
53     @SuppressWarnings("checkstyle:MemberName")
54     private final OperationState IDLE = new Idle();
55
56     private final RaftActor raftActor;
57
58     private final RaftActorContext raftContext;
59
60     private final Queue<ServerOperationContext<?>> pendingOperationsQueue = new ArrayDeque<>();
61
62     private OperationState currentOperationState = IDLE;
63
64     RaftActorServerConfigurationSupport(RaftActor raftActor) {
65         this.raftActor = raftActor;
66         this.raftContext = raftActor.getRaftActorContext();
67     }
68
69     boolean handleMessage(Object message, ActorRef sender) {
70         if (message instanceof AddServer) {
71             onAddServer((AddServer) message, sender);
72             return true;
73         } else if (message instanceof RemoveServer) {
74             onRemoveServer((RemoveServer) message, sender);
75             return true;
76         } else if (message instanceof ChangeServersVotingStatus) {
77             onChangeServersVotingStatus((ChangeServersVotingStatus) message, sender);
78             return true;
79         } else if (message instanceof ServerOperationTimeout) {
80             currentOperationState.onServerOperationTimeout((ServerOperationTimeout) message);
81             return true;
82         } else if (message instanceof UnInitializedFollowerSnapshotReply) {
83             currentOperationState.onUnInitializedFollowerSnapshotReply((UnInitializedFollowerSnapshotReply) message);
84             return true;
85         } else if (message instanceof ApplyState) {
86             return onApplyState((ApplyState) message);
87         } else if (message instanceof SnapshotComplete) {
88             currentOperationState.onSnapshotComplete();
89             return false;
90         } else {
91             return false;
92         }
93     }
94
95     void onNewLeader(String leaderId) {
96         currentOperationState.onNewLeader(leaderId);
97     }
98
99     private void onChangeServersVotingStatus(ChangeServersVotingStatus message, ActorRef sender) {
100         LOG.debug("{}: onChangeServersVotingStatus: {}, state: {}", raftContext.getId(), message,
101                 currentOperationState);
102
103         // The following check is a special case. Normally we fail an operation if there's no leader.
104         // Consider a scenario where one has 2 geographically-separated 3-node clusters, one a primary and
105         // the other a backup such that if the primary cluster is lost, the backup can take over. In this
106         // scenario, we have a logical 6-node cluster where the primary sub-cluster is configured as voting
107         // and the backup sub-cluster as non-voting such that the primary cluster can make progress without
108         // consensus from the backup cluster while still replicating to the backup. On fail-over to the backup,
109         // a request would be sent to a member of the backup cluster to flip the voting states, ie make the
110         // backup sub-cluster voting and the lost primary non-voting. However since the primary majority
111         // cluster is lost, there would be no leader to apply, persist and replicate the server config change.
112         // Therefore, if the local server is currently non-voting and is to be changed to voting and there is
113         // no current leader, we will try to elect a leader using the new server config in order to replicate
114         // the change and progress.
115         boolean localServerChangingToVoting = Boolean.TRUE.equals(message
116                 .getServerVotingStatusMap().get(raftActor.getRaftActorContext().getId()));
117         boolean hasNoLeader = raftActor.getLeaderId() == null;
118         if (localServerChangingToVoting && !raftContext.isVotingMember() && hasNoLeader) {
119             currentOperationState.onNewOperation(new ChangeServersVotingStatusContext(message, sender, true));
120         } else {
121             onNewOperation(new ChangeServersVotingStatusContext(message, sender, false));
122         }
123     }
124
125     private void onRemoveServer(RemoveServer removeServer, ActorRef sender) {
126         LOG.debug("{}: onRemoveServer: {}, state: {}", raftContext.getId(), removeServer, currentOperationState);
127         boolean isSelf = removeServer.getServerId().equals(raftContext.getId());
128         if (isSelf && !raftContext.hasFollowers()) {
129             sender.tell(new RemoveServerReply(ServerChangeStatus.NOT_SUPPORTED, raftActor.getLeaderId()),
130                     raftActor.getSelf());
131         } else if (!isSelf && !raftContext.getPeerIds().contains(removeServer.getServerId())) {
132             sender.tell(new RemoveServerReply(ServerChangeStatus.DOES_NOT_EXIST, raftActor.getLeaderId()),
133                     raftActor.getSelf());
134         } else {
135             String serverAddress = isSelf ? raftActor.self().path().toString() :
136                 raftContext.getPeerAddress(removeServer.getServerId());
137             onNewOperation(new RemoveServerContext(removeServer, serverAddress, sender));
138         }
139     }
140
141     private boolean onApplyState(ApplyState applyState) {
142         Payload data = applyState.getReplicatedLogEntry().getData();
143         if (data instanceof ServerConfigurationPayload) {
144             currentOperationState.onApplyState(applyState);
145             return true;
146         }
147
148         return false;
149     }
150
151     /**
152      * The algorithm for AddServer is as follows:
153      * <ul>
154      * <li>Add the new server as a peer.</li>
155      * <li>Add the new follower to the leader.</li>
156      * <li>If new server should be voting member</li>
157      * <ul>
158      *     <li>Initialize FollowerState to VOTING_NOT_INITIALIZED.</li>
159      *     <li>Initiate install snapshot to the new follower.</li>
160      *     <li>When install snapshot complete, mark the follower as VOTING and re-calculate majority vote count.</li>
161      * </ul>
162      * <li>Persist and replicate ServerConfigurationPayload with the new server list.</li>
163      * <li>On replication consensus, respond to caller with OK.</li>
164      * </ul>
165      * If the install snapshot times out after a period of 2 * election time out
166      * <ul>
167      *     <li>Remove the new server as a peer.</li>
168      *     <li>Remove the new follower from the leader.</li>
169      *     <li>Respond to caller with TIMEOUT.</li>
170      * </ul>
171      */
172     private void onAddServer(AddServer addServer, ActorRef sender) {
173         LOG.debug("{}: onAddServer: {}, state: {}", raftContext.getId(), addServer, currentOperationState);
174
175         onNewOperation(new AddServerContext(addServer, sender));
176     }
177
178     private void onNewOperation(ServerOperationContext<?> operationContext) {
179         if (raftActor.isLeader()) {
180             currentOperationState.onNewOperation(operationContext);
181         } else {
182             ActorSelection leader = raftActor.getLeader();
183             if (leader != null) {
184                 LOG.debug("{}: Not leader - forwarding to leader {}", raftContext.getId(), leader);
185                 leader.tell(operationContext.getOperation(), operationContext.getClientRequestor());
186             } else {
187                 LOG.debug("{}: No leader - returning NO_LEADER reply", raftContext.getId());
188                 operationContext.getClientRequestor().tell(operationContext.newReply(
189                         ServerChangeStatus.NO_LEADER, null), raftActor.self());
190             }
191         }
192     }
193
194     /**
195      * Interface for the initial state for a server operation.
196      */
197     private interface InitialOperationState {
198         void initiate();
199     }
200
201     /**
202      * Abstract base class for a server operation FSM state. Handles common behavior for all states.
203      */
204     private abstract class OperationState {
205         void onNewOperation(ServerOperationContext<?> operationContext) {
206             // We're currently processing another operation so queue it to be processed later.
207
208             LOG.debug("{}: Server operation already in progress - queueing {}", raftContext.getId(),
209                     operationContext.getOperation());
210
211             pendingOperationsQueue.add(operationContext);
212         }
213
214         void onServerOperationTimeout(ServerOperationTimeout timeout) {
215             LOG.debug("onServerOperationTimeout should not be called in state {}", this);
216         }
217
218         void onUnInitializedFollowerSnapshotReply(UnInitializedFollowerSnapshotReply reply) {
219             LOG.debug("onUnInitializedFollowerSnapshotReply was called in state {}", this);
220         }
221
222         void onApplyState(ApplyState applyState) {
223             LOG.debug("onApplyState was called in state {}", this);
224         }
225
226         void onSnapshotComplete() {
227
228         }
229
230         void onNewLeader(String newLeader) {
231         }
232
233         protected void persistNewServerConfiguration(ServerOperationContext<?> operationContext) {
234             raftContext.setDynamicServerConfigurationInUse();
235
236             ServerConfigurationPayload payload = raftContext.getPeerServerInfo(
237                     operationContext.includeSelfInNewConfiguration(raftActor));
238             LOG.debug("{}: New server configuration : {}", raftContext.getId(), payload.getServerConfig());
239
240             raftActor.persistData(operationContext.getClientRequestor(), operationContext.getContextId(),
241                     payload, false);
242
243             currentOperationState = new Persisting(operationContext, newTimer(new ServerOperationTimeout(
244                     operationContext.getLoggingContext())));
245
246             sendReply(operationContext, ServerChangeStatus.OK);
247         }
248
249         protected void operationComplete(ServerOperationContext<?> operationContext,
250                 @Nullable ServerChangeStatus replyStatus) {
251             if (replyStatus != null) {
252                 sendReply(operationContext, replyStatus);
253             }
254
255             operationContext.operationComplete(raftActor, replyStatus == null || replyStatus == ServerChangeStatus.OK);
256
257             changeToIdleState();
258         }
259
260         protected void changeToIdleState() {
261             currentOperationState = IDLE;
262
263             ServerOperationContext<?> nextOperation = pendingOperationsQueue.poll();
264             if (nextOperation != null) {
265                 RaftActorServerConfigurationSupport.this.onNewOperation(nextOperation);
266             }
267         }
268
269         protected void sendReply(ServerOperationContext<?> operationContext, ServerChangeStatus status) {
270             LOG.debug("{}: Returning {} for operation {}", raftContext.getId(), status,
271                     operationContext.getOperation());
272
273             operationContext.getClientRequestor().tell(operationContext.newReply(status, raftActor.getLeaderId()),
274                     raftActor.self());
275         }
276
277         Cancellable newTimer(Object message) {
278             return newTimer(raftContext.getConfigParams().getElectionTimeOutInterval().$times(2), message);
279         }
280
281         Cancellable newTimer(FiniteDuration timeout, Object message) {
282             return raftContext.getActorSystem().scheduler().scheduleOnce(
283                     timeout, raftContext.getActor(), message,
284                             raftContext.getActorSystem().dispatcher(), raftContext.getActor());
285         }
286
287         @Override
288         public String toString() {
289             return getClass().getSimpleName();
290         }
291     }
292
293     /**
294      * The state when no server operation is in progress. It immediately initiates new server operations.
295      */
296     private final class Idle extends OperationState {
297         @Override
298         public void onNewOperation(ServerOperationContext<?> operationContext) {
299             operationContext.newInitialOperationState(RaftActorServerConfigurationSupport.this).initiate();
300         }
301
302         @Override
303         public void onApplyState(ApplyState applyState) {
304             // Noop - we override b/c ApplyState is called normally for followers in the idle state.
305         }
306     }
307
308     /**
309      * The state when a new server configuration is being persisted and replicated.
310      */
311     private final class Persisting extends OperationState {
312         private final ServerOperationContext<?> operationContext;
313         private final Cancellable timer;
314         private boolean timedOut = false;
315
316         Persisting(ServerOperationContext<?> operationContext, Cancellable timer) {
317             this.operationContext = operationContext;
318             this.timer = timer;
319         }
320
321         @Override
322         public void onApplyState(ApplyState applyState) {
323             // Sanity check - we could get an ApplyState from a previous operation that timed out so make
324             // sure it's meant for us.
325             if (operationContext.getContextId().equals(applyState.getIdentifier())) {
326                 LOG.info("{}: {} has been successfully replicated to a majority of followers", raftContext.getId(),
327                         applyState.getReplicatedLogEntry().getData());
328
329                 timer.cancel();
330                 operationComplete(operationContext, null);
331             }
332         }
333
334         @Override
335         public void onServerOperationTimeout(ServerOperationTimeout timeout) {
336             LOG.warn("{}: Timeout occured while replicating the new server configuration for {}", raftContext.getId(),
337                     timeout.getLoggingContext());
338
339             timedOut = true;
340
341             // Fail any pending operations
342             ServerOperationContext<?> nextOperation = pendingOperationsQueue.poll();
343             while (nextOperation != null) {
344                 sendReply(nextOperation, ServerChangeStatus.PRIOR_REQUEST_CONSENSUS_TIMEOUT);
345                 nextOperation = pendingOperationsQueue.poll();
346             }
347         }
348
349         @Override
350         public void onNewOperation(ServerOperationContext<?> newOperationContext) {
351             if (timedOut) {
352                 sendReply(newOperationContext, ServerChangeStatus.PRIOR_REQUEST_CONSENSUS_TIMEOUT);
353             } else {
354                 super.onNewOperation(newOperationContext);
355             }
356         }
357     }
358
359     /**
360      * Abstract base class for an AddServer operation state.
361      */
362     private abstract class AddServerState extends OperationState {
363         private final AddServerContext addServerContext;
364
365         AddServerState(AddServerContext addServerContext) {
366             this.addServerContext = addServerContext;
367         }
368
369         AddServerContext getAddServerContext() {
370             return addServerContext;
371         }
372
373         Cancellable newInstallSnapshotTimer() {
374             return newTimer(new ServerOperationTimeout(addServerContext.getOperation().getNewServerId()));
375         }
376
377         void handleInstallSnapshotTimeout(ServerOperationTimeout timeout) {
378             String serverId = timeout.getLoggingContext();
379
380             LOG.debug("{}: handleInstallSnapshotTimeout for new server {}", raftContext.getId(), serverId);
381
382             // cleanup
383             raftContext.removePeer(serverId);
384
385             boolean isLeader = raftActor.isLeader();
386             if (isLeader) {
387                 AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
388                 leader.removeFollower(serverId);
389             }
390
391             operationComplete(getAddServerContext(), isLeader ? ServerChangeStatus.TIMEOUT
392                     : ServerChangeStatus.NO_LEADER);
393         }
394
395     }
396
397     /**
398      * The initial state for the AddServer operation. It adds the new follower as a peer and initiates
399      * snapshot capture, if necessary.
400      */
401     private final class InitialAddServerState extends AddServerState implements InitialOperationState {
402         InitialAddServerState(AddServerContext addServerContext) {
403             super(addServerContext);
404         }
405
406         @Override
407         public void initiate() {
408             final AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
409             AddServer addServer = getAddServerContext().getOperation();
410
411             LOG.debug("{}: Initiating {}", raftContext.getId(), addServer);
412
413             if (raftContext.getPeerInfo(addServer.getNewServerId()) != null) {
414                 operationComplete(getAddServerContext(), ServerChangeStatus.ALREADY_EXISTS);
415                 return;
416             }
417
418             VotingState votingState = addServer.isVotingMember() ? VotingState.VOTING_NOT_INITIALIZED :
419                     VotingState.NON_VOTING;
420             raftContext.addToPeers(addServer.getNewServerId(), addServer.getNewServerAddress(), votingState);
421
422             leader.addFollower(addServer.getNewServerId());
423
424             if (votingState == VotingState.VOTING_NOT_INITIALIZED) {
425                 // schedule the install snapshot timeout timer
426                 Cancellable installSnapshotTimer = newInstallSnapshotTimer();
427                 if (leader.initiateCaptureSnapshot(addServer.getNewServerId())) {
428                     LOG.debug("{}: Initiating capture snapshot for new server {}", raftContext.getId(),
429                             addServer.getNewServerId());
430
431                     currentOperationState = new InstallingSnapshot(getAddServerContext(), installSnapshotTimer);
432                 } else {
433                     LOG.debug("{}: Snapshot already in progress - waiting for completion", raftContext.getId());
434
435                     currentOperationState = new WaitingForPriorSnapshotComplete(getAddServerContext(),
436                             installSnapshotTimer);
437                 }
438             } else {
439                 LOG.debug("{}: New follower is non-voting - directly persisting new server configuration",
440                         raftContext.getId());
441
442                 persistNewServerConfiguration(getAddServerContext());
443             }
444         }
445     }
446
447     /**
448      * The AddServer operation state for when the catch-up snapshot is being installed. It handles successful
449      * reply or timeout.
450      */
451     private final class InstallingSnapshot extends AddServerState {
452         private final Cancellable installSnapshotTimer;
453
454         InstallingSnapshot(AddServerContext addServerContext, Cancellable installSnapshotTimer) {
455             super(addServerContext);
456             this.installSnapshotTimer = Preconditions.checkNotNull(installSnapshotTimer);
457         }
458
459         @Override
460         public void onServerOperationTimeout(ServerOperationTimeout timeout) {
461             handleInstallSnapshotTimeout(timeout);
462
463             LOG.warn("{}: Timeout occured for new server {} while installing snapshot", raftContext.getId(),
464                     timeout.getLoggingContext());
465         }
466
467         @Override
468         public void onUnInitializedFollowerSnapshotReply(UnInitializedFollowerSnapshotReply reply) {
469             LOG.debug("{}: onUnInitializedFollowerSnapshotReply: {}", raftContext.getId(), reply);
470
471             String followerId = reply.getFollowerId();
472
473             // Sanity check to guard against receiving an UnInitializedFollowerSnapshotReply from a prior
474             // add server operation that timed out.
475             if (getAddServerContext().getOperation().getNewServerId().equals(followerId) && raftActor.isLeader()) {
476                 AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
477                 raftContext.getPeerInfo(followerId).setVotingState(VotingState.VOTING);
478                 leader.updateMinReplicaCount();
479
480                 persistNewServerConfiguration(getAddServerContext());
481
482                 installSnapshotTimer.cancel();
483             } else {
484                 LOG.debug("{}: Dropping UnInitializedFollowerSnapshotReply for server {}: {}",
485                         raftContext.getId(), followerId,
486                         !raftActor.isLeader() ? "not leader" : "server Id doesn't match");
487             }
488         }
489     }
490
491     /**
492      * The AddServer operation state for when there is a snapshot already in progress. When the current
493      * snapshot completes, it initiates an install snapshot.
494      */
495     private final class WaitingForPriorSnapshotComplete extends AddServerState {
496         private final Cancellable snapshotTimer;
497
498         WaitingForPriorSnapshotComplete(AddServerContext addServerContext, Cancellable snapshotTimer) {
499             super(addServerContext);
500             this.snapshotTimer = Preconditions.checkNotNull(snapshotTimer);
501         }
502
503         @Override
504         public void onSnapshotComplete() {
505             LOG.debug("{}: onSnapshotComplete", raftContext.getId());
506
507             if (!raftActor.isLeader()) {
508                 LOG.debug("{}: No longer the leader", raftContext.getId());
509                 return;
510             }
511
512             AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
513             if (leader.initiateCaptureSnapshot(getAddServerContext().getOperation().getNewServerId())) {
514                 LOG.debug("{}: Initiating capture snapshot for new server {}", raftContext.getId(),
515                         getAddServerContext().getOperation().getNewServerId());
516
517                 currentOperationState = new InstallingSnapshot(getAddServerContext(),
518                         newInstallSnapshotTimer());
519
520                 snapshotTimer.cancel();
521             }
522         }
523
524         @Override
525         public void onServerOperationTimeout(ServerOperationTimeout timeout) {
526             handleInstallSnapshotTimeout(timeout);
527
528             LOG.warn("{}: Timeout occured for new server {} while waiting for prior snapshot to complete",
529                     raftContext.getId(), timeout.getLoggingContext());
530         }
531     }
532
533     private static final class ServerOperationContextIdentifier
534             extends AbstractUUIDIdentifier<ServerOperationContextIdentifier> {
535         private static final long serialVersionUID = 1L;
536
537         ServerOperationContextIdentifier() {
538             super(UUID.randomUUID());
539         }
540     }
541
542     /**
543      * Stores context information for a server operation.
544      *
545      * @param <T> the operation type
546      */
547     private abstract static class ServerOperationContext<T> {
548         private final T operation;
549         private final ActorRef clientRequestor;
550         private final Identifier contextId;
551
552         ServerOperationContext(T operation, ActorRef clientRequestor) {
553             this.operation = operation;
554             this.clientRequestor = clientRequestor;
555             contextId = new ServerOperationContextIdentifier();
556         }
557
558         Identifier getContextId() {
559             return contextId;
560         }
561
562         T getOperation() {
563             return operation;
564         }
565
566         ActorRef getClientRequestor() {
567             return clientRequestor;
568         }
569
570         void operationComplete(RaftActor raftActor, boolean succeeded) {
571         }
572
573         boolean includeSelfInNewConfiguration(RaftActor raftActor) {
574             return true;
575         }
576
577         abstract Object newReply(ServerChangeStatus status, String leaderId);
578
579         abstract InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support);
580
581         abstract String getLoggingContext();
582     }
583
584     /**
585      * Stores context information for an AddServer operation.
586      */
587     private static class AddServerContext extends ServerOperationContext<AddServer> {
588         AddServerContext(AddServer addServer, ActorRef clientRequestor) {
589             super(addServer, clientRequestor);
590         }
591
592         @Override
593         Object newReply(ServerChangeStatus status, String leaderId) {
594             return new AddServerReply(status, leaderId);
595         }
596
597         @Override
598         InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support) {
599             return support.new InitialAddServerState(this);
600         }
601
602         @Override
603         String getLoggingContext() {
604             return getOperation().getNewServerId();
605         }
606     }
607
608     private abstract class RemoveServerState extends OperationState {
609         private final RemoveServerContext removeServerContext;
610
611         protected RemoveServerState(RemoveServerContext removeServerContext) {
612             this.removeServerContext = Preconditions.checkNotNull(removeServerContext);
613
614         }
615
616         public RemoveServerContext getRemoveServerContext() {
617             return removeServerContext;
618         }
619     }
620
621     private final class InitialRemoveServerState extends RemoveServerState implements InitialOperationState {
622
623         protected InitialRemoveServerState(RemoveServerContext removeServerContext) {
624             super(removeServerContext);
625         }
626
627         @Override
628         public void initiate() {
629             String serverId = getRemoveServerContext().getOperation().getServerId();
630             raftContext.removePeer(serverId);
631             ((AbstractLeader)raftActor.getCurrentBehavior()).removeFollower(serverId);
632
633             persistNewServerConfiguration(getRemoveServerContext());
634         }
635     }
636
637     private static class RemoveServerContext extends ServerOperationContext<RemoveServer> {
638         private final String peerAddress;
639
640         RemoveServerContext(RemoveServer operation, String peerAddress, ActorRef clientRequestor) {
641             super(operation, clientRequestor);
642             this.peerAddress = peerAddress;
643         }
644
645         @Override
646         Object newReply(ServerChangeStatus status, String leaderId) {
647             return new RemoveServerReply(status, leaderId);
648         }
649
650         @Override
651         InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support) {
652             return support.new InitialRemoveServerState(this);
653         }
654
655         @Override
656         void operationComplete(RaftActor raftActor, boolean succeeded) {
657             if (peerAddress != null) {
658                 raftActor.context().actorSelection(peerAddress).tell(
659                         new ServerRemoved(getOperation().getServerId()), raftActor.getSelf());
660             }
661         }
662
663         @Override
664         boolean includeSelfInNewConfiguration(RaftActor raftActor) {
665             return !getOperation().getServerId().equals(raftActor.getId());
666         }
667
668         @Override
669         String getLoggingContext() {
670             return getOperation().getServerId();
671         }
672     }
673
674     private static class ChangeServersVotingStatusContext extends ServerOperationContext<ChangeServersVotingStatus> {
675         private final boolean tryToElectLeader;
676
677         ChangeServersVotingStatusContext(ChangeServersVotingStatus convertMessage, ActorRef clientRequestor,
678                 boolean tryToElectLeader) {
679             super(convertMessage, clientRequestor);
680             this.tryToElectLeader = tryToElectLeader;
681         }
682
683         @Override
684         InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support) {
685             return support.new ChangeServersVotingStatusState(this, tryToElectLeader);
686         }
687
688         @Override
689         Object newReply(ServerChangeStatus status, String leaderId) {
690             return new ServerChangeReply(status, leaderId);
691         }
692
693         @Override
694         void operationComplete(final RaftActor raftActor, boolean succeeded) {
695             // If this leader changed to non-voting we need to step down as leader so we'll try to transfer
696             // leadership.
697             boolean localServerChangedToNonVoting = Boolean.FALSE.equals(getOperation()
698                     .getServerVotingStatusMap().get(raftActor.getRaftActorContext().getId()));
699             if (succeeded && localServerChangedToNonVoting) {
700                 LOG.debug("Leader changed to non-voting - trying leadership transfer");
701                 raftActor.becomeNonVoting();
702             }
703         }
704
705         @Override
706         String getLoggingContext() {
707             return getOperation().toString();
708         }
709     }
710
711     private class ChangeServersVotingStatusState extends OperationState implements InitialOperationState {
712         private final ChangeServersVotingStatusContext changeVotingStatusContext;
713         private final boolean tryToElectLeader;
714
715         ChangeServersVotingStatusState(ChangeServersVotingStatusContext changeVotingStatusContext,
716                 boolean tryToElectLeader) {
717             this.changeVotingStatusContext = changeVotingStatusContext;
718             this.tryToElectLeader = tryToElectLeader;
719         }
720
721         @Override
722         public void initiate() {
723             LOG.debug("Initiating ChangeServersVotingStatusState");
724
725             if (tryToElectLeader) {
726                 initiateLocalLeaderElection();
727             } else if (updateLocalPeerInfo()) {
728                 persistNewServerConfiguration(changeVotingStatusContext);
729             }
730         }
731
732         private void initiateLocalLeaderElection() {
733             LOG.debug("{}: Sending local ElectionTimeout to start leader election", raftContext.getId());
734
735             ServerConfigurationPayload previousServerConfig = raftContext.getPeerServerInfo(true);
736             if (!updateLocalPeerInfo()) {
737                 return;
738             }
739
740             raftContext.getActor().tell(TimeoutNow.INSTANCE, raftContext.getActor());
741
742             currentOperationState = new WaitingForLeaderElected(changeVotingStatusContext, previousServerConfig);
743         }
744
745         private boolean updateLocalPeerInfo() {
746             List<ServerInfo> newServerInfoList = newServerInfoList();
747
748             // Check if new voting state would leave us with no voting members.
749             boolean atLeastOneVoting = false;
750             for (ServerInfo info: newServerInfoList) {
751                 if (info.isVoting()) {
752                     atLeastOneVoting = true;
753                     break;
754                 }
755             }
756
757             if (!atLeastOneVoting) {
758                 operationComplete(changeVotingStatusContext, ServerChangeStatus.INVALID_REQUEST);
759                 return false;
760             }
761
762             raftContext.updatePeerIds(new ServerConfigurationPayload(newServerInfoList));
763             if (raftActor.getCurrentBehavior() instanceof AbstractLeader) {
764                 AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
765                 leader.updateMinReplicaCount();
766             }
767
768             return true;
769         }
770
771         private List<ServerInfo> newServerInfoList() {
772             Map<String, Boolean> serverVotingStatusMap = changeVotingStatusContext.getOperation()
773                     .getServerVotingStatusMap();
774             List<ServerInfo> newServerInfoList = new ArrayList<>();
775             for (String peerId: raftContext.getPeerIds()) {
776                 newServerInfoList.add(new ServerInfo(peerId, serverVotingStatusMap.containsKey(peerId)
777                         ? serverVotingStatusMap.get(peerId) : raftContext.getPeerInfo(peerId).isVoting()));
778             }
779
780             newServerInfoList.add(new ServerInfo(raftContext.getId(), serverVotingStatusMap.containsKey(
781                     raftContext.getId()) ? serverVotingStatusMap.get(raftContext.getId())
782                             : raftContext.isVotingMember()));
783
784             return newServerInfoList;
785         }
786     }
787
788     private class WaitingForLeaderElected extends OperationState {
789         private final ServerConfigurationPayload previousServerConfig;
790         private final ChangeServersVotingStatusContext operationContext;
791         private final Cancellable timer;
792
793         WaitingForLeaderElected(ChangeServersVotingStatusContext operationContext,
794                 ServerConfigurationPayload previousServerConfig) {
795             this.operationContext = operationContext;
796             this.previousServerConfig = previousServerConfig;
797
798             timer = newTimer(raftContext.getConfigParams().getElectionTimeOutInterval(),
799                     new ServerOperationTimeout(operationContext.getLoggingContext()));
800         }
801
802         @Override
803         void onNewLeader(String newLeader) {
804             if (newLeader == null) {
805                 return;
806             }
807
808             LOG.debug("{}: New leader {} elected", raftContext.getId(), newLeader);
809
810             timer.cancel();
811
812             if (raftActor.isLeader()) {
813                 persistNewServerConfiguration(operationContext);
814             } else {
815                 // Edge case - some other node became leader so forward the operation.
816                 LOG.debug("{}: Forwarding {} to new leader", raftContext.getId(), operationContext.getOperation());
817
818                 // Revert the local server config change.
819                 raftContext.updatePeerIds(previousServerConfig);
820
821                 changeToIdleState();
822                 RaftActorServerConfigurationSupport.this.onNewOperation(operationContext);
823             }
824         }
825
826         @Override
827         void onServerOperationTimeout(ServerOperationTimeout timeout) {
828             LOG.warn("{}: Leader election timed out - cannot apply operation {}",
829                     raftContext.getId(), timeout.getLoggingContext());
830
831             // Revert the local server config change.
832             raftContext.updatePeerIds(previousServerConfig);
833             raftActor.initializeBehavior();
834
835             tryToForwardOperationToAnotherServer();
836         }
837
838         private void tryToForwardOperationToAnotherServer() {
839             Collection<String> serversVisited = new HashSet<>(operationContext.getOperation().getServersVisited());
840
841             LOG.debug("{}: tryToForwardOperationToAnotherServer - servers already visited {}", raftContext.getId(),
842                     serversVisited);
843
844             serversVisited.add(raftContext.getId());
845
846             // Try to find another whose state is being changed from non-voting to voting and that we haven't
847             // tried yet.
848             Map<String, Boolean> serverVotingStatusMap = operationContext.getOperation().getServerVotingStatusMap();
849             ActorSelection forwardToPeerActor = null;
850             for (Map.Entry<String, Boolean> e: serverVotingStatusMap.entrySet()) {
851                 Boolean isVoting = e.getValue();
852                 String serverId = e.getKey();
853                 PeerInfo peerInfo = raftContext.getPeerInfo(serverId);
854                 if (isVoting && peerInfo != null && !peerInfo.isVoting() && !serversVisited.contains(serverId)) {
855                     ActorSelection actor = raftContext.getPeerActorSelection(serverId);
856                     if (actor != null) {
857                         forwardToPeerActor = actor;
858                         break;
859                     }
860                 }
861             }
862
863             if (forwardToPeerActor != null) {
864                 LOG.debug("{}: Found server {} to forward to", raftContext.getId(), forwardToPeerActor);
865
866                 forwardToPeerActor.tell(new ChangeServersVotingStatus(serverVotingStatusMap, serversVisited),
867                         operationContext.getClientRequestor());
868                 changeToIdleState();
869             } else {
870                 operationComplete(operationContext, ServerChangeStatus.NO_LEADER);
871             }
872         }
873     }
874
875     static class ServerOperationTimeout {
876         private final String loggingContext;
877
878         ServerOperationTimeout(String loggingContext) {
879             this.loggingContext = Preconditions.checkNotNull(loggingContext, "loggingContext should not be null");
880         }
881
882         String getLoggingContext() {
883             return loggingContext;
884         }
885     }
886 }