Change ReplicatedLogImplEntry to Externalizable proxy pattern
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / ReplicationAndSnapshotsIntegrationTest.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
12 import akka.persistence.SaveSnapshotSuccess;
13 import com.google.common.collect.ImmutableMap;
14 import java.util.List;
15 import org.junit.Test;
16 import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
17 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
18 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
19 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
20 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
21 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
22 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
23 import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
24 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
25 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
26 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
27
28 /**
29  * Tests replication and snapshots end-to-end using real RaftActors and behavior communication.
30  *
31  * @author Thomas Pantelis
32  */
33 public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorIntegrationTest {
34
35     private List<SimpleReplicatedLogEntry> origLeaderJournal;
36
37     private MockPayload recoveredPayload0;
38     private MockPayload recoveredPayload1;
39     private MockPayload recoveredPayload2;
40     private MockPayload payload3;
41     private MockPayload payload4;
42     private MockPayload payload5;
43     private MockPayload payload6;
44     private MockPayload payload7;
45
46     @Test
47     public void runTest() throws Exception {
48         testLog.info("testReplicationAndSnapshots starting");
49
50         // Setup the persistent journal for the leader. We'll start up with 3 journal log entries (one less
51         // than the snapshotBatchCount).
52         long seqId = 1;
53         InMemoryJournal.addEntry(leaderId, seqId++, new UpdateElectionTerm(initialTerm, leaderId));
54         recoveredPayload0 = new MockPayload("zero");
55         InMemoryJournal.addEntry(leaderId, seqId++, new SimpleReplicatedLogEntry(0, initialTerm, recoveredPayload0));
56         recoveredPayload1 = new MockPayload("one");
57         InMemoryJournal.addEntry(leaderId, seqId++, new SimpleReplicatedLogEntry(1, initialTerm, recoveredPayload1));
58         recoveredPayload2 = new MockPayload("two");
59         InMemoryJournal.addEntry(leaderId, seqId++, new SimpleReplicatedLogEntry(2, initialTerm, recoveredPayload2));
60         InMemoryJournal.addEntry(leaderId, seqId++, new ApplyJournalEntries(2));
61
62         origLeaderJournal = InMemoryJournal.get(leaderId, SimpleReplicatedLogEntry.class);
63
64         // Create the leader and 2 follower actors and verify initial syncing of the followers after leader
65         // persistence recovery.
66
67         DefaultConfigParamsImpl followerConfigParams = newFollowerConfigParams();
68         followerConfigParams.setSnapshotBatchCount(snapshotBatchCount);
69         follower1Actor = newTestRaftActor(follower1Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
70                 follower2Id, testActorPath(follower2Id)), followerConfigParams);
71
72         follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
73                 follower1Id, testActorPath(follower1Id)), followerConfigParams);
74
75         peerAddresses = ImmutableMap.<String, String>builder()
76                 .put(follower1Id, follower1Actor.path().toString())
77                 .put(follower2Id, follower2Actor.path().toString()).build();
78
79         leaderConfigParams = newLeaderConfigParams();
80         leaderActor = newTestRaftActor(leaderId, peerAddresses, leaderConfigParams);
81
82         follower1CollectorActor = follower1Actor.underlyingActor().collectorActor();
83         follower2CollectorActor = follower2Actor.underlyingActor().collectorActor();
84         leaderCollectorActor = leaderActor.underlyingActor().collectorActor();
85
86         leaderContext = leaderActor.underlyingActor().getRaftActorContext();
87
88         verifyLeaderRecoveryAndInitialization();
89
90         testFirstSnapshot();
91
92         testSubsequentReplications();
93
94         testSecondSnapshot();
95
96         testLeaderReinstatement();
97
98         testLog.info("testReplicationAndSnapshots ending");
99     }
100
101     /**
102      * Verify the expected leader is elected as the leader and verify initial syncing of the followers
103      * from the leader's persistence recovery.
104      */
105     void verifyLeaderRecoveryAndInitialization() {
106         testLog.info("verifyLeaderRecoveryAndInitialization starting");
107
108         waitUntilLeader(leaderActor);
109
110         currentTerm = leaderContext.getTermInformation().getCurrentTerm();
111         assertEquals("Current term > " + initialTerm, true, currentTerm > initialTerm);
112
113         leader = leaderActor.underlyingActor().getCurrentBehavior();
114
115         // The followers should receive AppendEntries for each leader log entry that was recovered from
116         // persistence and apply each one.
117         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(
118                 follower1CollectorActor, ApplyState.class, 3);
119         verifyApplyState(applyStates.get(0), null, null, initialTerm, 0, recoveredPayload0);
120         verifyApplyState(applyStates.get(1), null, null, initialTerm, 1, recoveredPayload1);
121         verifyApplyState(applyStates.get(2), null, null, initialTerm, 2, recoveredPayload2);
122
123         // Verify follower 1 applies a log entry for at least the last entry index.
124         verifyApplyJournalEntries(follower1CollectorActor, 2);
125
126         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 3);
127         verifyApplyState(applyStates.get(0), null, null, initialTerm, 0, recoveredPayload0);
128         verifyApplyState(applyStates.get(1), null, null, initialTerm, 1, recoveredPayload1);
129         verifyApplyState(applyStates.get(2), null, null, initialTerm, 2, recoveredPayload2);
130
131         // Verify follower 1]2 applies a log entry for at least the last entry index.
132         verifyApplyJournalEntries(follower2CollectorActor, 2);
133
134         MessageCollectorActor.clearMessages(leaderCollectorActor);
135         MessageCollectorActor.clearMessages(follower1CollectorActor);
136         MessageCollectorActor.clearMessages(follower2CollectorActor);
137
138         // The leader should have performed fake snapshots due to the follower's AppendEntriesReplies and
139         // trimmed the in-memory log so that only the last entry remains.
140         assertEquals("Leader snapshot term", initialTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
141         assertEquals("Leader snapshot index", 1, leaderContext.getReplicatedLog().getSnapshotIndex());
142         assertEquals("Leader journal log size", 1, leaderContext.getReplicatedLog().size());
143         assertEquals("Leader journal last index", 2, leaderContext.getReplicatedLog().lastIndex());
144         assertEquals("Leader commit index", 2, leaderContext.getCommitIndex());
145         assertEquals("Leader last applied", 2, leaderContext.getLastApplied());
146         assertEquals("Leader replicatedToAllIndex", 1, leader.getReplicatedToAllIndex());
147
148         // Verify the follower's persisted journal log.
149         verifyPersistedJournal(follower1Id, origLeaderJournal);
150         verifyPersistedJournal(follower2Id, origLeaderJournal);
151
152         MessageCollectorActor.clearMessages(leaderCollectorActor);
153         MessageCollectorActor.clearMessages(follower1CollectorActor);
154         MessageCollectorActor.clearMessages(follower2CollectorActor);
155
156         testLog.info("verifyLeaderRecoveryAndInitialization ending");
157     }
158
159     /**
160      * Send a payload to the TestRaftActor to persist and replicate. Since snapshotBatchCount is set to
161      * 4 and we already have 3 entries in the journal log, this should initiate a snapshot. In this
162      * scenario, the follower consensus and application of state is delayed until after the snapshot
163      * completes.
164      */
165     private void testFirstSnapshot() throws Exception {
166         testLog.info("testFirstSnapshot starting");
167
168         expSnapshotState.add(recoveredPayload0);
169         expSnapshotState.add(recoveredPayload1);
170         expSnapshotState.add(recoveredPayload2);
171
172         // Delay the consensus by temporarily dropping the AppendEntries to both followers.
173         follower1Actor.underlyingActor().startDropMessages(AppendEntries.class);
174         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
175
176         // Send the payload.
177         payload3 = sendPayloadData(leaderActor, "three");
178
179         // Wait for snapshot complete.
180         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
181
182         // The snapshot index should not be advanced nor the log trimmed because replicatedToAllIndex
183         // is behind due the followers not being replicated yet via AppendEntries.
184         assertEquals("Leader snapshot term", initialTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
185         assertEquals("Leader snapshot index", 1, leaderContext.getReplicatedLog().getSnapshotIndex());
186         assertEquals("Leader journal log size", 2, leaderContext.getReplicatedLog().size());
187         assertEquals("Leader journal last index", 3, leaderContext.getReplicatedLog().lastIndex());
188
189         // Verify the persisted snapshot in the leader. This should reflect the advanced snapshot index as
190         // the last applied log entry (2) even though the leader hasn't yet advanced its cached snapshot index.
191         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
192         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
193         verifySnapshot("Persisted", persistedSnapshots.get(0), initialTerm, 2, currentTerm, 3);
194         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
195         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
196         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 3, payload3);
197
198         // The leader's persisted journal log should be cleared since we snapshotted.
199         List<SimpleReplicatedLogEntry> persistedLeaderJournal =
200                 InMemoryJournal.get(leaderId, SimpleReplicatedLogEntry.class);
201         assertEquals("Persisted journal log size", 0, persistedLeaderJournal.size());
202
203         // Allow AppendEntries to both followers to proceed. This should catch up the followers and cause a
204         // "fake" snapshot in the leader to advance the snapshot index to 2. Also the state should be applied
205         // in all members (via ApplyState).
206         follower1Actor.underlyingActor().stopDropMessages(AppendEntries.class);
207         follower2Actor.underlyingActor().stopDropMessages(AppendEntries.class);
208
209         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
210         verifyApplyState(applyState, leaderCollectorActor, payload3.toString(), currentTerm, 3, payload3);
211
212         verifyApplyJournalEntries(leaderCollectorActor, 3);
213
214         assertEquals("Leader commit index", 3, leaderContext.getCommitIndex());
215
216         applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class);
217         verifyApplyState(applyState, null, null, currentTerm, 3, payload3);
218
219         verifyApplyJournalEntries(follower1CollectorActor, 3);
220
221         applyState = MessageCollectorActor.expectFirstMatching(follower2CollectorActor, ApplyState.class);
222         verifyApplyState(applyState, null, null, currentTerm, 3, payload3);
223
224         verifyApplyJournalEntries(follower2CollectorActor, 3);
225
226         assertEquals("Leader snapshot term", initialTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
227         assertEquals("Leader snapshot index", 2, leaderContext.getReplicatedLog().getSnapshotIndex());
228         assertEquals("Leader journal log size", 1, leaderContext.getReplicatedLog().size());
229         assertEquals("Leader commit index", 3, leaderContext.getCommitIndex());
230         assertEquals("Leader last applied", 3, leaderContext.getLastApplied());
231         assertEquals("Leader replicatedToAllIndex", 2, leader.getReplicatedToAllIndex());
232
233         // The followers should also snapshot so verify.
234
235         MessageCollectorActor.expectFirstMatching(follower1CollectorActor, SaveSnapshotSuccess.class);
236         persistedSnapshots = InMemorySnapshotStore.getSnapshots(follower1Id, Snapshot.class);
237         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
238         // The last applied index in the snapshot may or may not be the last log entry depending on
239         // timing so to avoid intermittent test failures, we'll just verify the snapshot's last term/index.
240         assertEquals("Follower1 Snapshot getLastTerm", currentTerm, persistedSnapshots.get(0).getLastTerm());
241         assertEquals("Follower1 Snapshot getLastIndex", 3, persistedSnapshots.get(0).getLastIndex());
242
243         MessageCollectorActor.expectFirstMatching(follower2CollectorActor, SaveSnapshotSuccess.class);
244
245         MessageCollectorActor.clearMessages(leaderCollectorActor);
246         MessageCollectorActor.clearMessages(follower1CollectorActor);
247         MessageCollectorActor.clearMessages(follower2CollectorActor);
248
249         testLog.info("testFirstSnapshot ending");
250     }
251
252     /**
253      * Send 3 more payload instances and verify they get applied by all members.
254      */
255     private void testSubsequentReplications() {
256         testLog.info("testSubsequentReplications starting");
257
258         payload4 = sendPayloadData(leaderActor, "four");
259         payload5 = sendPayloadData(leaderActor, "five");
260         payload6 = sendPayloadData(leaderActor, "six");
261
262         // Verify the leader applies the states.
263         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 3);
264         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
265         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload5.toString(), currentTerm, 5, payload5);
266         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload6.toString(), currentTerm, 6, payload6);
267
268         // Verify the leader applies a log entry for at least the last entry index.
269         verifyApplyJournalEntries(leaderCollectorActor, 6);
270
271         // The leader should have performed fake snapshots due to the follower's AppendEntriesReplies and
272         // trimmed the in-memory log so that only the last entry remains.
273         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
274         assertEquals("Leader snapshot index", 5, leaderContext.getReplicatedLog().getSnapshotIndex());
275         assertEquals("Leader journal log size", 1, leaderContext.getReplicatedLog().size());
276         assertEquals("Leader journal last index", 6, leaderContext.getReplicatedLog().lastIndex());
277         assertEquals("Leader commit index", 6, leaderContext.getCommitIndex());
278         assertEquals("Leader last applied", 6, leaderContext.getLastApplied());
279         assertEquals("Leader replicatedToAllIndex", 5, leader.getReplicatedToAllIndex());
280
281         // Verify follower 1 applies the states.
282         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 3);
283         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
284         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
285         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
286
287         // Verify follower 1 applies a log entry for at least the last entry index.
288         verifyApplyJournalEntries(follower1CollectorActor, 6);
289
290         // Verify follower 2 applies the states.
291         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 3);
292         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
293         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
294         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
295
296         // Verify follower 2 applies a log entry for at least the last entry index.
297         verifyApplyJournalEntries(follower2CollectorActor, 6);
298
299         MessageCollectorActor.clearMessages(leaderCollectorActor);
300
301         testLog.info("testSubsequentReplications ending");
302     }
303
304     /**
305      * Send one more payload to trigger another snapshot. In this scenario, we delay the snapshot until
306      * consensus occurs and the leader applies the state.
307      */
308     private void testSecondSnapshot() throws Exception {
309         testLog.info("testSecondSnapshot starting");
310
311         expSnapshotState.add(payload3);
312         expSnapshotState.add(payload4);
313         expSnapshotState.add(payload5);
314         expSnapshotState.add(payload6);
315
316         // Delay the CaptureSnapshot message to the leader actor.
317         leaderActor.underlyingActor().startDropMessages(CaptureSnapshotReply.class);
318
319         // Send the payload.
320         payload7 = sendPayloadData(leaderActor, "seven");
321
322         // Capture the CaptureSnapshotReply message so we can send it later.
323         final CaptureSnapshotReply captureSnapshotReply = MessageCollectorActor.expectFirstMatching(
324                 leaderCollectorActor, CaptureSnapshotReply.class);
325
326         // Wait for the state to be applied in the leader.
327         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
328         verifyApplyState(applyState, leaderCollectorActor, payload7.toString(), currentTerm, 7, payload7);
329
330         // At this point the leader has applied the new state but the cached snapshot index should not be
331         // advanced by a "fake" snapshot because we're in the middle of a snapshot. We'll wait for at least
332         // one more heartbeat AppendEntriesReply to ensure this does not occur.
333         MessageCollectorActor.clearMessages(leaderCollectorActor);
334         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
335
336         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
337         assertEquals("Leader snapshot index", 5, leaderContext.getReplicatedLog().getSnapshotIndex());
338         assertEquals("Leader journal log size", 2, leaderContext.getReplicatedLog().size());
339         assertEquals("Leader journal last index", 7, leaderContext.getReplicatedLog().lastIndex());
340         assertEquals("Leader commit index", 7, leaderContext.getCommitIndex());
341         assertEquals("Leader last applied", 7, leaderContext.getLastApplied());
342         assertEquals("Leader replicatedToAllIndex", 5, leader.getReplicatedToAllIndex());
343
344         // Now deliver the CaptureSnapshotReply.
345         leaderActor.underlyingActor().stopDropMessages(CaptureSnapshotReply.class);
346         leaderActor.tell(captureSnapshotReply, leaderActor);
347
348         // Wait for snapshot complete.
349         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
350
351         // Wait for another heartbeat AppendEntriesReply. This should cause a "fake" snapshot to advance the
352         // snapshot index and trimmed the log since we're no longer in a snapshot.
353         MessageCollectorActor.clearMessages(leaderCollectorActor);
354         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
355         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
356         assertEquals("Leader snapshot index", 6, leaderContext.getReplicatedLog().getSnapshotIndex());
357         assertEquals("Leader journal log size", 1, leaderContext.getReplicatedLog().size());
358         assertEquals("Leader journal last index", 7, leaderContext.getReplicatedLog().lastIndex());
359         assertEquals("Leader commit index", 7, leaderContext.getCommitIndex());
360
361         // Verify the persisted snapshot. This should reflect the snapshot index as the last applied
362         // log entry (7) and shouldn't contain any unapplied entries as we capture persisted the snapshot data
363         // when the snapshot is created (ie when the CaptureSnapshot is processed).
364         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
365         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
366         verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 6, currentTerm, 7);
367         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
368         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
369         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 7, payload7);
370
371         // The leader's persisted journal log should be cleared since we did a snapshot.
372         List<SimpleReplicatedLogEntry> persistedLeaderJournal = InMemoryJournal.get(
373                 leaderId, SimpleReplicatedLogEntry.class);
374         assertEquals("Persisted journal log size", 0, persistedLeaderJournal.size());
375
376         // Verify the followers apply all 4 new log entries.
377         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor,
378                 ApplyState.class, 4);
379         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
380         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
381         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
382         verifyApplyState(applyStates.get(3), null, null, currentTerm, 7, payload7);
383
384         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 4);
385         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
386         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
387         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
388         verifyApplyState(applyStates.get(3), null, null, currentTerm, 7, payload7);
389
390         // Verify the follower's snapshot index has also advanced. (after another AppendEntries heartbeat
391         // to be safe).
392
393         MessageCollectorActor.clearMessages(follower1CollectorActor);
394         MessageCollectorActor.expectFirstMatching(follower1CollectorActor, AppendEntries.class);
395         follower1Context = follower1Actor.underlyingActor().getRaftActorContext();
396         assertEquals("Follower 1 snapshot term", currentTerm, follower1Context.getReplicatedLog().getSnapshotTerm());
397         assertEquals("Follower 1 snapshot index", 6, follower1Context.getReplicatedLog().getSnapshotIndex());
398         assertEquals("Follower 1 journal log size", 1, follower1Context.getReplicatedLog().size());
399         assertEquals("Follower 1 journal last index", 7, follower1Context.getReplicatedLog().lastIndex());
400         assertEquals("Follower 1 commit index", 7, follower1Context.getCommitIndex());
401
402         MessageCollectorActor.clearMessages(follower2CollectorActor);
403         MessageCollectorActor.expectFirstMatching(follower2CollectorActor, AppendEntries.class);
404         follower2Context = follower2Actor.underlyingActor().getRaftActorContext();
405         assertEquals("Follower 2 snapshot term", currentTerm, follower2Context.getReplicatedLog().getSnapshotTerm());
406         assertEquals("Follower 2 snapshot index", 6, follower2Context.getReplicatedLog().getSnapshotIndex());
407         assertEquals("Follower 2 journal log size", 1, follower2Context.getReplicatedLog().size());
408         assertEquals("Follower 2 journal last index", 7, follower2Context.getReplicatedLog().lastIndex());
409         assertEquals("Follower 2 commit index", 7, follower2Context.getCommitIndex());
410
411         expSnapshotState.add(payload7);
412
413         testLog.info("testSecondSnapshot ending");
414     }
415
416     /**
417      * Kill the leader actor, reinstate it and verify the recovered journal.
418      */
419     private void testLeaderReinstatement() {
420         testLog.info("testLeaderReinstatement starting");
421
422         killActor(leaderActor);
423
424         leaderActor = newTestRaftActor(leaderId, peerAddresses, leaderConfigParams);
425
426         leaderActor.underlyingActor().waitForRecoveryComplete();
427
428         leaderContext = leaderActor.underlyingActor().getRaftActorContext();
429
430         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
431         assertEquals("Leader snapshot index", 6, leaderContext.getReplicatedLog().getSnapshotIndex());
432         assertEquals("Leader journal log size", 1, leaderContext.getReplicatedLog().size());
433         assertEquals("Leader journal last index", 7, leaderContext.getReplicatedLog().lastIndex());
434         assertEquals("Leader commit index", 7, leaderContext.getCommitIndex());
435         assertEquals("Leader last applied", 7, leaderContext.getLastApplied());
436         verifyReplicatedLogEntry(leaderContext.getReplicatedLog().last(), currentTerm, 7, payload7);
437
438         testLog.info("testLeaderReinstatement ending");
439     }
440 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.