Bug 2187: Code cleanup and refactoring
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / RaftActorServerConfigurationSupportTest.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 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 akka.actor.ActorRef;
14 import akka.actor.Props;
15 import akka.actor.UntypedActor;
16 import akka.dispatch.Dispatchers;
17 import akka.testkit.JavaTestKit;
18 import akka.testkit.TestActorRef;
19 import com.google.common.base.Optional;
20 import com.google.common.collect.ImmutableMap;
21 import com.google.common.collect.Maps;
22 import com.google.common.collect.Sets;
23 import java.util.Collections;
24 import java.util.List;
25 import java.util.Map;
26 //import java.util.List;
27 import java.util.concurrent.TimeUnit;
28 import org.junit.After;
29 import org.junit.Before;
30 import org.junit.Test;
31 import org.opendaylight.controller.cluster.DataPersistenceProvider;
32 import org.opendaylight.controller.cluster.NonPersistentDataProvider;
33 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
34 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
35 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
36 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader;
37 import org.opendaylight.controller.cluster.raft.behaviors.Follower;
38 import org.opendaylight.controller.cluster.raft.behaviors.Leader;
39 import org.opendaylight.controller.cluster.raft.messages.AddServer;
40 import org.opendaylight.controller.cluster.raft.messages.AddServerReply;
41 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
42 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
43 import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
44 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
45 import org.opendaylight.controller.cluster.raft.utils.ForwardMessageToBehaviorActor;
46 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
47 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
48 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
49 import org.slf4j.Logger;
50 import org.slf4j.LoggerFactory;
51 import scala.concurrent.duration.FiniteDuration;
52
53 /**
54  * Unit tests for RaftActorServerConfigurationSupport.
55  *
56  * @author Thomas Pantelis
57  */
58 public class RaftActorServerConfigurationSupportTest extends AbstractActorTest {
59     static final String LEADER_ID = "leader";
60     static final String FOLLOWER_ID = "follower";
61     static final String NEW_SERVER_ID = "new-server";
62     static final String NEW_SERVER_ID2 = "new-server2";
63     private static final Logger LOG = LoggerFactory.getLogger(RaftActorServerConfigurationSupportTest.class);
64     private static final DataPersistenceProvider NO_PERSISTENCE = new NonPersistentDataProvider();
65
66     private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
67
68     private final TestActorRef<ForwardMessageToBehaviorActor> followerActor = actorFactory.createTestActor(
69             Props.create(ForwardMessageToBehaviorActor.class).withDispatcher(Dispatchers.DefaultDispatcherId()),
70             actorFactory.generateActorId(FOLLOWER_ID));
71
72     private TestActorRef<MockNewFollowerRaftActor> newFollowerRaftActor;
73     private TestActorRef<MessageCollectorActor> newFollowerCollectorActor;
74     private RaftActorContext newFollowerActorContext;
75
76     private final JavaTestKit testKit = new JavaTestKit(getSystem());
77
78     @Before
79     public void setup() {
80         InMemoryJournal.clear();
81         InMemorySnapshotStore.clear();
82
83         DefaultConfigParamsImpl configParams = newFollowerConfigParams();
84
85         newFollowerCollectorActor = actorFactory.createTestActor(
86                 MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()),
87                 actorFactory.generateActorId(NEW_SERVER_ID + "Collector"));
88         newFollowerRaftActor = actorFactory.createTestActor(MockNewFollowerRaftActor.props(
89                 configParams, newFollowerCollectorActor).withDispatcher(Dispatchers.DefaultDispatcherId()),
90                 actorFactory.generateActorId(NEW_SERVER_ID));
91
92         try {
93             newFollowerActorContext = newFollowerRaftActor.underlyingActor().getRaftActorContext();
94         } catch (Exception e) {
95             newFollowerActorContext = newFollowerRaftActor.underlyingActor().getRaftActorContext();
96         }
97     }
98
99     private DefaultConfigParamsImpl newFollowerConfigParams() {
100         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
101         configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
102         configParams.setElectionTimeoutFactor(100000);
103         configParams.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
104         return configParams;
105     }
106
107     @After
108     public void tearDown() throws Exception {
109         actorFactory.close();
110     }
111
112     @Test
113     public void testAddServerWithExistingFollower() throws Exception {
114         RaftActorContext followerActorContext = newFollowerContext(FOLLOWER_ID, followerActor);
115         followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(
116                 0, 3, 1).build());
117         followerActorContext.setCommitIndex(2);
118         followerActorContext.setLastApplied(2);
119
120         Follower follower = new Follower(followerActorContext);
121         followerActor.underlyingActor().setBehavior(follower);
122
123         TestActorRef<MockLeaderRaftActor> leaderActor = actorFactory.createTestActor(
124                 MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()),
125                         followerActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()),
126                 actorFactory.generateActorId(LEADER_ID));
127
128         // Expect initial heartbeat from the leader.
129         expectFirstMatching(followerActor, AppendEntries.class);
130         clearMessages(followerActor);
131
132         MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor();
133
134         leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef());
135
136         // Leader should install snapshot - capture and verify ApplySnapshot contents
137
138         ApplySnapshot applySnapshot = expectFirstMatching(newFollowerCollectorActor, ApplySnapshot.class);
139         List<Object> snapshotState = (List<Object>) MockRaftActor.toObject(applySnapshot.getSnapshot().getState());
140         assertEquals("Snapshot state", snapshotState, leaderRaftActor.getState());
141
142         AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
143         assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus());
144         assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint());
145
146         // Verify ServerConfigurationPayload entry in leader's log
147
148         RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext();
149         assertEquals("Leader journal last index", 3, leaderActorContext.getReplicatedLog().lastIndex());
150         assertEquals("Leader commit index", 3, leaderActorContext.getCommitIndex());
151         assertEquals("Leader last applied index", 3, leaderActorContext.getLastApplied());
152         verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), LEADER_ID, FOLLOWER_ID, NEW_SERVER_ID);
153
154         // Verify ServerConfigurationPayload entry in both followers
155
156         assertEquals("Follower journal last index", 3, followerActorContext.getReplicatedLog().lastIndex());
157         verifyServerConfigurationPayloadEntry(followerActorContext.getReplicatedLog(), LEADER_ID, FOLLOWER_ID, NEW_SERVER_ID);
158
159         assertEquals("New follower journal last index", 3, newFollowerActorContext.getReplicatedLog().lastIndex());
160         verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), LEADER_ID, FOLLOWER_ID, NEW_SERVER_ID);
161
162         // Verify new server config was applied in both followers
163
164         assertEquals("Follower peers", Sets.newHashSet(LEADER_ID, NEW_SERVER_ID),
165                 followerActorContext.getPeerAddresses().keySet());
166
167         assertEquals("New follower peers", Sets.newHashSet(LEADER_ID, FOLLOWER_ID),
168                 newFollowerActorContext.getPeerAddresses().keySet());
169
170         expectFirstMatching(newFollowerCollectorActor, ApplyState.class);
171         expectFirstMatching(followerActor, ApplyState.class);
172
173         assertEquals("Follower commit index", 3, followerActorContext.getCommitIndex());
174         assertEquals("Follower last applied index", 3, followerActorContext.getLastApplied());
175         assertEquals("New follower commit index", 3, newFollowerActorContext.getCommitIndex());
176         assertEquals("New follower last applied index", 3, newFollowerActorContext.getLastApplied());
177     }
178
179     @Test
180     public void testAddServerWithNoExistingFollower() throws Exception {
181         RaftActorContext initialActorContext = new MockRaftActorContext();
182         initialActorContext.setCommitIndex(1);
183         initialActorContext.setLastApplied(1);
184         initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(
185                 0, 2, 1).build());
186
187         TestActorRef<MockLeaderRaftActor> leaderActor = actorFactory.createTestActor(
188                 MockLeaderRaftActor.props(ImmutableMap.<String, String>of(),
189                         initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()),
190                 actorFactory.generateActorId(LEADER_ID));
191
192         MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor();
193         RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext();
194
195         leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef());
196
197         // Leader should install snapshot - capture and verify ApplySnapshot contents
198
199         ApplySnapshot applySnapshot = expectFirstMatching(newFollowerCollectorActor, ApplySnapshot.class);
200         List<Object> snapshotState = (List<Object>) MockRaftActor.toObject(applySnapshot.getSnapshot().getState());
201         assertEquals("Snapshot state", snapshotState, leaderRaftActor.getState());
202
203         AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
204         assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus());
205         assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint());
206
207         // Verify ServerConfigurationPayload entry in leader's log
208
209         assertEquals("Leader journal last index", 2, leaderActorContext.getReplicatedLog().lastIndex());
210         assertEquals("Leader commit index", 2, leaderActorContext.getCommitIndex());
211         assertEquals("Leader last applied index", 2, leaderActorContext.getLastApplied());
212         verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), LEADER_ID, NEW_SERVER_ID);
213
214         // Verify ServerConfigurationPayload entry in the new follower
215
216         expectFirstMatching(newFollowerCollectorActor, ApplyState.class);
217         assertEquals("New follower journal last index", 2, newFollowerActorContext.getReplicatedLog().lastIndex());
218         verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), LEADER_ID, NEW_SERVER_ID);
219
220         // Verify new server config was applied in the new follower
221
222         assertEquals("New follower peers", Sets.newHashSet(LEADER_ID),
223                 newFollowerActorContext.getPeerAddresses().keySet());
224     }
225
226     @Test
227     public void testAddServersAsNonVoting() throws Exception {
228         RaftActorContext initialActorContext = new MockRaftActorContext();
229         initialActorContext.setCommitIndex(-1);
230         initialActorContext.setLastApplied(-1);
231         initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
232
233         TestActorRef<MockLeaderRaftActor> leaderActor = actorFactory.createTestActor(
234                 MockLeaderRaftActor.props(ImmutableMap.<String, String>of(),
235                         initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()),
236                 actorFactory.generateActorId(LEADER_ID));
237
238         MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor();
239         RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext();
240
241         leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), false), testKit.getRef());
242
243         AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
244         assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus());
245         assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint());
246
247         // Verify ServerConfigurationPayload entry in leader's log
248
249         assertEquals("Leader journal last index", 0, leaderActorContext.getReplicatedLog().lastIndex());
250         assertEquals("Leader commit index", 0, leaderActorContext.getCommitIndex());
251         assertEquals("Leader last applied index", 0, leaderActorContext.getLastApplied());
252         verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), LEADER_ID, NEW_SERVER_ID);
253
254         // Verify ServerConfigurationPayload entry in the new follower
255
256         expectFirstMatching(newFollowerCollectorActor, ApplyState.class);
257         assertEquals("New follower journal last index", 0, newFollowerActorContext.getReplicatedLog().lastIndex());
258         verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), LEADER_ID, NEW_SERVER_ID);
259
260         // Verify new server config was applied in the new follower
261
262         assertEquals("New follower peers", Sets.newHashSet(LEADER_ID),
263                 newFollowerActorContext.getPeerAddresses().keySet());
264
265         MessageCollectorActor.assertNoneMatching(newFollowerCollectorActor, InstallSnapshot.class, 500);
266
267         // Add another non-voting server.
268
269         RaftActorContext follower2ActorContext = newFollowerContext(NEW_SERVER_ID2, followerActor);
270         Follower newFollower2 = new Follower(follower2ActorContext);
271         followerActor.underlyingActor().setBehavior(newFollower2);
272
273         leaderActor.tell(new AddServer(NEW_SERVER_ID2, followerActor.path().toString(), false), testKit.getRef());
274
275         addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
276         assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus());
277         assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint());
278
279         assertEquals("Leader journal last index", 1, leaderActorContext.getReplicatedLog().lastIndex());
280         assertEquals("Leader commit index", 1, leaderActorContext.getCommitIndex());
281         assertEquals("Leader last applied index", 1, leaderActorContext.getLastApplied());
282         verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(),
283                 LEADER_ID, NEW_SERVER_ID, NEW_SERVER_ID2);
284     }
285
286     @Test
287     public void testAddServerWithOperationInProgress() throws Exception {
288         RaftActorContext initialActorContext = new MockRaftActorContext();
289         initialActorContext.setCommitIndex(-1);
290         initialActorContext.setLastApplied(-1);
291         initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
292
293         TestActorRef<MockLeaderRaftActor> leaderActor = actorFactory.createTestActor(
294                 MockLeaderRaftActor.props(ImmutableMap.<String, String>of(),
295                         initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()),
296                 actorFactory.generateActorId(LEADER_ID));
297
298         MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor();
299         RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext();
300
301         RaftActorContext follower2ActorContext = newFollowerContext(NEW_SERVER_ID2, followerActor);
302         Follower newFollower2 = new Follower(follower2ActorContext);
303         followerActor.underlyingActor().setBehavior(newFollower2);
304
305         MockNewFollowerRaftActor newFollowerRaftActorInstance = newFollowerRaftActor.underlyingActor();
306         newFollowerRaftActorInstance.setDropMessageOfType(InstallSnapshot.SERIALIZABLE_CLASS);
307
308         leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef());
309
310         // Wait for leader's install snapshot and capture it
311
312         Object installSnapshot = expectFirstMatching(newFollowerCollectorActor, InstallSnapshot.class);
313
314         JavaTestKit testKit2 = new JavaTestKit(getSystem());
315         leaderActor.tell(new AddServer(NEW_SERVER_ID2, followerActor.path().toString(), false), testKit2.getRef());
316
317         newFollowerRaftActorInstance.setDropMessageOfType(null);
318         newFollowerRaftActor.tell(installSnapshot, leaderActor);
319
320         AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
321         assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus());
322
323         addServerReply = testKit2.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
324         assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus());
325
326         // Verify ServerConfigurationPayload entries in leader's log
327
328         assertEquals("Leader journal last index", 1, leaderActorContext.getReplicatedLog().lastIndex());
329         assertEquals("Leader commit index", 1, leaderActorContext.getCommitIndex());
330         assertEquals("Leader last applied index", 1, leaderActorContext.getLastApplied());
331         verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(),
332                 LEADER_ID, NEW_SERVER_ID, NEW_SERVER_ID2);
333
334         // Verify ServerConfigurationPayload entry in the new follower
335
336         MessageCollectorActor.expectMatching(newFollowerCollectorActor, ApplyState.class, 2);
337
338         assertEquals("New follower peers", Sets.newHashSet(LEADER_ID, NEW_SERVER_ID2),
339                newFollowerActorContext.getPeerAddresses().keySet());
340     }
341
342     @Test
343     public void testAddServerWithInstallSnapshotTimeout() throws Exception {
344         newFollowerRaftActor.underlyingActor().setDropMessageOfType(InstallSnapshot.SERIALIZABLE_CLASS);
345
346         RaftActorContext initialActorContext = new MockRaftActorContext();
347         initialActorContext.setCommitIndex(-1);
348         initialActorContext.setLastApplied(-1);
349         initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
350
351         TestActorRef<MockLeaderRaftActor> leaderActor = actorFactory.createTestActor(
352                 MockLeaderRaftActor.props(ImmutableMap.<String, String>of(),
353                         initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()),
354                 actorFactory.generateActorId(LEADER_ID));
355
356         MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor();
357         RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext();
358         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(1);
359
360         leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef());
361
362         AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
363         assertEquals("getStatus", ServerChangeStatus.TIMEOUT, addServerReply.getStatus());
364
365         assertEquals("Leader peers size", 0, leaderActorContext.getPeerAddresses().keySet().size());
366         assertEquals("Leader followers size", 0,
367                 ((AbstractLeader)leaderRaftActor.getCurrentBehavior()).getFollowerIds().size());
368     }
369
370     @Test
371     public void testAddServerWithNoLeader() {
372         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
373         configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
374
375         TestActorRef<MockRaftActor> noLeaderActor = actorFactory.createTestActor(
376                 MockRaftActor.props(LEADER_ID, ImmutableMap.<String,String>of(FOLLOWER_ID, followerActor.path().toString()),
377                         Optional.<ConfigParams>of(configParams), NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()),
378                 actorFactory.generateActorId(LEADER_ID));
379         noLeaderActor.underlyingActor().waitForInitializeBehaviorComplete();
380
381         noLeaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef());
382         AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class);
383         assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus());
384     }
385
386     @Test
387     public void testAddServerForwardedToLeader() {
388         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
389         configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
390
391         TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
392                 MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()),
393                 actorFactory.generateActorId(LEADER_ID));
394
395         TestActorRef<MockRaftActor> followerRaftActor = actorFactory.createTestActor(
396                 MockRaftActor.props(FOLLOWER_ID, ImmutableMap.<String,String>of(LEADER_ID, leaderActor.path().toString()),
397                         Optional.<ConfigParams>of(configParams), NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()),
398                 actorFactory.generateActorId(FOLLOWER_ID));
399         followerRaftActor.underlyingActor().waitForInitializeBehaviorComplete();
400
401         followerRaftActor.tell(new AppendEntries(1, LEADER_ID, 0, 1, Collections.<ReplicatedLogEntry>emptyList(),
402                 -1, -1, (short)0), leaderActor);
403
404         followerRaftActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef());
405         expectFirstMatching(leaderActor, AddServer.class);
406     }
407
408     private void verifyServerConfigurationPayloadEntry(ReplicatedLog log, String... cNew) {
409         ReplicatedLogEntry logEntry = log.get(log.lastIndex());
410         assertEquals("Last log entry payload class", ServerConfigurationPayload.class, logEntry.getData().getClass());
411         ServerConfigurationPayload payload = (ServerConfigurationPayload)logEntry.getData();
412         assertEquals("getNewServerConfig", Sets.newHashSet(cNew), Sets.newHashSet(payload.getNewServerConfig()));
413     }
414
415     private RaftActorContext newFollowerContext(String id, TestActorRef<? extends UntypedActor> actor) {
416         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
417         configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
418         configParams.setElectionTimeoutFactor(100000);
419         ElectionTermImpl termInfo = new ElectionTermImpl(NO_PERSISTENCE, id, LOG);
420         termInfo.update(1, LEADER_ID);
421         RaftActorContext followerActorContext = new RaftActorContextImpl(actor, actor.underlyingActor().getContext(),
422                 id, termInfo, -1, -1,
423                 ImmutableMap.of(LEADER_ID, ""), configParams, NO_PERSISTENCE, LOG);
424         followerActorContext.setCommitIndex(-1);
425         followerActorContext.setLastApplied(-1);
426         followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
427
428         return followerActorContext;
429     }
430
431     public static class MockLeaderRaftActor extends MockRaftActor {
432         public MockLeaderRaftActor(Map<String, String> peerAddresses, ConfigParams config,
433                 RaftActorContext fromContext) {
434             super(LEADER_ID, peerAddresses, Optional.of(config), NO_PERSISTENCE);
435
436             RaftActorContext context = getRaftActorContext();
437             for(int i = 0; i < fromContext.getReplicatedLog().size(); i++) {
438                 ReplicatedLogEntry entry = fromContext.getReplicatedLog().get(i);
439                 getState().add(entry.getData());
440                 context.getReplicatedLog().append(entry);
441             }
442
443             context.setCommitIndex(fromContext.getCommitIndex());
444             context.setLastApplied(fromContext.getLastApplied());
445             context.getTermInformation().update(fromContext.getTermInformation().getCurrentTerm(),
446                     fromContext.getTermInformation().getVotedFor());
447         }
448
449         @Override
450         protected void initializeBehavior() {
451             changeCurrentBehavior(new Leader(getRaftActorContext()));
452             initializeBehaviorComplete.countDown();
453         }
454
455         @Override
456         public void createSnapshot(ActorRef actorRef) {
457             try {
458                 actorRef.tell(new CaptureSnapshotReply(RaftActorTest.fromObject(getState()).toByteArray()), actorRef);
459             } catch (Exception e) {
460                 LOG.error("createSnapshot failed", e);
461             }
462         }
463
464         static Props props(Map<String, String> peerAddresses, RaftActorContext fromContext) {
465             DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
466             configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
467             configParams.setElectionTimeoutFactor(10);
468             return Props.create(MockLeaderRaftActor.class, peerAddresses, configParams, fromContext);
469         }
470     }
471
472     public static class MockNewFollowerRaftActor extends MockRaftActor {
473         private final TestActorRef<MessageCollectorActor> collectorActor;
474         private volatile Class<?> dropMessageOfType;
475
476         public MockNewFollowerRaftActor(ConfigParams config, TestActorRef<MessageCollectorActor> collectorActor) {
477             super(NEW_SERVER_ID, Maps.<String, String>newHashMap(), Optional.of(config), null);
478             this.collectorActor = collectorActor;
479         }
480
481         void setDropMessageOfType(Class<?> dropMessageOfType) {
482             this.dropMessageOfType = dropMessageOfType;
483         }
484
485         @Override
486         public void handleCommand(Object message) {
487             if(dropMessageOfType == null || !dropMessageOfType.equals(message.getClass())) {
488                 super.handleCommand(message);
489             }
490
491             collectorActor.tell(message, getSender());
492         }
493
494         static Props props(ConfigParams config, TestActorRef<MessageCollectorActor> collectorActor) {
495             return Props.create(MockNewFollowerRaftActor.class, config, collectorActor);
496         }
497     }
498 }