2 * Copyright (c) 2015 Brocade Communications 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
8 package org.opendaylight.controller.cluster.raft;
10 import static org.junit.Assert.assertEquals;
11 import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor.clearMessages;
12 import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor.expectFirstMatching;
13 import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor.expectMatching;
14 import akka.actor.ActorRef;
15 import akka.actor.Props;
16 import akka.pattern.Patterns;
17 import akka.testkit.TestActorRef;
18 import com.google.common.collect.ImmutableMap;
19 import java.util.Collections;
20 import java.util.Iterator;
21 import java.util.List;
22 import java.util.concurrent.TimeUnit;
23 import org.junit.Test;
24 import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
25 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
26 import org.opendaylight.controller.cluster.raft.base.messages.LeaderTransitioning;
27 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
28 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
29 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
30 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
31 import scala.concurrent.Await;
32 import scala.concurrent.Future;
33 import scala.concurrent.duration.FiniteDuration;
36 * Tests leadership transfer end-to-end.
38 * @author Thomas Pantelis
40 public class LeadershipTransferIntegrationTest extends AbstractRaftActorIntegrationTest {
42 private final String follower3Id = factory.generateActorId("follower");
43 private TestActorRef<MessageCollectorActor> leaderNotifierActor;
44 private TestActorRef<MessageCollectorActor> follower1NotifierActor;
45 private TestActorRef<MessageCollectorActor> follower2NotifierActor;
46 private TestActorRef<MessageCollectorActor> follower3NotifierActor;
47 private TestActorRef<TestRaftActor> follower3Actor;
48 private ActorRef follower3CollectorActor;
51 public void testLeaderTransferOnShutDown() throws Throwable {
52 testLog.info("testLeaderTransferOnShutDown starting");
56 sendPayloadWithFollower2Lagging();
58 sendShutDownToLeaderAndVerifyLeadershipTransferToFollower1();
60 sendShutDown(follower2Actor);
62 testLog.info("testLeaderTransferOnShutDown ending");
65 private void sendShutDown(ActorRef actor) throws Exception {
66 testLog.info("sendShutDown for {} starting", actor.path());
68 FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
69 Future<Boolean> stopFuture = Patterns.gracefulStop(actor, duration, Shutdown.INSTANCE);
71 Boolean stopped = Await.result(stopFuture, duration);
72 assertEquals("Stopped", Boolean.TRUE, stopped);
74 testLog.info("sendShutDown for {} ending", actor.path());
77 private void sendShutDownToLeaderAndVerifyLeadershipTransferToFollower1() throws Throwable {
78 testLog.info("sendShutDownToLeaderAndVerifyLeadershipTransferToFollower1 starting");
80 clearMessages(leaderNotifierActor);
81 clearMessages(follower1NotifierActor);
82 clearMessages(follower2NotifierActor);
83 clearMessages(follower3NotifierActor);
85 // Simulate a delay for follower2 in receiving the LeaderTransitioning message with null leader id.
86 final TestRaftActor follower2Instance = follower2Actor.underlyingActor();
87 follower2Instance.startDropMessages(LeaderTransitioning.class);
89 FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
90 Future<Boolean> stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE);
92 verifyRaftState(follower1Actor, RaftState.Leader);
94 Boolean stopped = Await.result(stopFuture, duration);
95 assertEquals("Stopped", Boolean.TRUE, stopped);
97 // Re-enable LeaderTransitioning messages to follower2.
98 final LeaderTransitioning leaderTransitioning = expectFirstMatching(follower2CollectorActor,
99 LeaderTransitioning.class);
100 follower2Instance.stopDropMessages(LeaderTransitioning.class);
102 follower2Instance.stopDropMessages(AppendEntries.class);
103 ApplyState applyState = expectFirstMatching(follower2CollectorActor, ApplyState.class);
104 assertEquals("Apply sate index", 0, applyState.getReplicatedLogEntry().getIndex());
106 // Now send the LeaderTransitioning to follower2 after it has received AppendEntries from the new leader.
107 follower2Actor.tell(leaderTransitioning, ActorRef.noSender());
109 verifyLeaderStateChangedMessages(leaderNotifierActor, null, follower1Id);
110 verifyLeaderStateChangedMessages(follower1NotifierActor, null, follower1Id);
111 // follower2 should only get 1 LeaderStateChanged with the new leaderId - the LeaderTransitioning message
112 // should not generate a LeaderStateChanged with null leaderId since it arrived after the new leaderId was set.
113 verifyLeaderStateChangedMessages(follower2NotifierActor, follower1Id);
114 verifyLeaderStateChangedMessages(follower3NotifierActor, null, follower1Id);
116 testLog.info("sendShutDownToLeaderAndVerifyLeadershipTransferToFollower1 ending");
119 private void sendPayloadWithFollower2Lagging() {
120 testLog.info("sendPayloadWithFollower2Lagging starting");
122 follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
124 sendPayloadData(leaderActor, "zero");
126 expectFirstMatching(leaderCollectorActor, ApplyState.class);
127 expectFirstMatching(follower1CollectorActor, ApplyState.class);
128 expectFirstMatching(follower3CollectorActor, ApplyState.class);
130 testLog.info("sendPayloadWithFollower2Lagging ending");
133 private void createRaftActors() {
134 testLog.info("createRaftActors starting");
136 follower1NotifierActor = factory.createTestActor(Props.create(MessageCollectorActor.class),
137 factory.generateActorId(follower1Id + "-notifier"));
138 follower1Actor = newTestRaftActor(follower1Id, TestRaftActor.newBuilder().peerAddresses(
139 ImmutableMap.of(leaderId, testActorPath(leaderId), follower2Id, testActorPath(follower2Id),
140 follower3Id, testActorPath(follower3Id))).
141 config(newFollowerConfigParams()).roleChangeNotifier(follower1NotifierActor));
143 follower2NotifierActor = factory.createTestActor(Props.create(MessageCollectorActor.class),
144 factory.generateActorId(follower2Id + "-notifier"));
145 follower2Actor = newTestRaftActor(follower2Id,TestRaftActor.newBuilder().peerAddresses(
146 ImmutableMap.of(leaderId, testActorPath(leaderId), follower1Id, follower1Actor.path().toString(),
147 follower3Id, testActorPath(follower3Id))).
148 config(newFollowerConfigParams()).roleChangeNotifier(follower2NotifierActor));
150 follower3NotifierActor = factory.createTestActor(Props.create(MessageCollectorActor.class),
151 factory.generateActorId(follower3Id + "-notifier"));
152 follower3Actor = newTestRaftActor(follower3Id,TestRaftActor.newBuilder().peerAddresses(
153 ImmutableMap.of(leaderId, testActorPath(leaderId), follower1Id, follower1Actor.path().toString(),
154 follower2Id, follower2Actor.path().toString())).
155 config(newFollowerConfigParams()).roleChangeNotifier(follower3NotifierActor));
157 peerAddresses = ImmutableMap.<String, String>builder().
158 put(follower1Id, follower1Actor.path().toString()).
159 put(follower2Id, follower2Actor.path().toString()).
160 put(follower3Id, follower3Actor.path().toString()).build();
162 leaderConfigParams = newLeaderConfigParams();
163 leaderConfigParams.setElectionTimeoutFactor(3);
164 leaderNotifierActor = factory.createTestActor(Props.create(MessageCollectorActor.class),
165 factory.generateActorId(leaderId + "-notifier"));
166 leaderActor = newTestRaftActor(leaderId, TestRaftActor.newBuilder().peerAddresses(peerAddresses).
167 config(leaderConfigParams).roleChangeNotifier(leaderNotifierActor));
169 follower1CollectorActor = follower1Actor.underlyingActor().collectorActor();
170 follower2CollectorActor = follower2Actor.underlyingActor().collectorActor();
171 follower3CollectorActor = follower3Actor.underlyingActor().collectorActor();
172 leaderCollectorActor = leaderActor.underlyingActor().collectorActor();
174 leaderContext = leaderActor.underlyingActor().getRaftActorContext();
175 leaderContext.getPeerInfo(follower3Id).setVotingState(VotingState.NON_VOTING);
177 waitUntilLeader(leaderActor);
179 testLog.info("createRaftActors starting");
182 private static void verifyRaftState(ActorRef raftActor, final RaftState expState) {
183 verifyRaftState(raftActor, rs -> assertEquals("getRaftState", expState.toString(), rs.getRaftState()));
186 private void verifyLeaderStateChangedMessages(TestActorRef<MessageCollectorActor> notifierActor,
187 String... expLeaderIds) {
188 List<LeaderStateChanged> leaderStateChanges = expectMatching(notifierActor, LeaderStateChanged.class,
189 expLeaderIds.length);
191 Collections.reverse(leaderStateChanges);
192 Iterator<LeaderStateChanged> actual = leaderStateChanges.iterator();
193 for (int i = expLeaderIds.length - 1; i >= 0; i--) {
194 assertEquals("getLeaderId", expLeaderIds[i], actual.next().getLeaderId());
199 public void testLeaderTransferAborted() throws Throwable {
200 testLog.info("testLeaderTransferAborted starting");
204 leaderActor.underlyingActor().startDropMessages(AppendEntriesReply.class);
206 sendShutDown(leaderActor);
208 verifyRaftState(follower1Actor, RaftState.Follower);
209 verifyRaftState(follower2Actor, RaftState.Follower);
210 verifyRaftState(follower3Actor, RaftState.Follower);
212 testLog.info("testLeaderTransferOnShutDown ending");
216 public void testLeaderTransferSkippedOnShutdownWithNoFollowers() throws Throwable {
217 testLog.info("testLeaderTransferSkippedOnShutdownWithNoFollowers starting");
219 leaderActor = newTestRaftActor(leaderId, TestRaftActor.newBuilder().config(newLeaderConfigParams()));
220 waitUntilLeader(leaderActor);
222 sendShutDown(leaderActor);
224 testLog.info("testLeaderTransferSkippedOnShutdownWithNoFollowers ending");