2 * Copyright (c) 2016 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;
12 import akka.actor.ActorRef;
13 import akka.dispatch.Dispatchers;
14 import akka.testkit.TestActorRef;
15 import com.google.common.base.Optional;
16 import com.google.common.collect.ImmutableMap;
17 import com.google.common.collect.Sets;
18 import com.google.common.io.ByteSource;
19 import com.google.common.util.concurrent.Uninterruptibles;
20 import java.io.OutputStream;
21 import java.io.Serializable;
22 import java.util.Arrays;
23 import java.util.List;
24 import java.util.concurrent.TimeUnit;
25 import java.util.function.Consumer;
26 import org.apache.commons.lang3.SerializationUtils;
27 import org.junit.After;
28 import org.junit.Before;
29 import org.junit.Test;
30 import org.opendaylight.controller.cluster.raft.MockRaftActor.MockSnapshotState;
31 import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
32 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
33 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
34 import org.opendaylight.controller.cluster.raft.persisted.ByteState;
35 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
36 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
37 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
38 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
39 import org.opendaylight.controller.cluster.raft.persisted.Snapshot.State;
40 import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
41 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
42 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
43 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
44 import org.slf4j.Logger;
45 import org.slf4j.LoggerFactory;
48 * Unit tests for migrated messages on recovery.
50 * @author Thomas Pantelis
52 public class MigratedMessagesTest extends AbstractActorTest {
53 static final Logger TEST_LOG = LoggerFactory.getLogger(MigratedMessagesTest.class);
55 private TestActorFactory factory;
59 factory = new TestActorFactory(getSystem());
63 public void tearDown() throws Exception {
65 InMemoryJournal.clear();
66 InMemorySnapshotStore.clear();
70 public void testNoSnapshotAfterStartupWithNoMigratedMessages() {
71 TEST_LOG.info("testNoSnapshotAfterStartupWithNoMigratedMessages starting");
72 String id = factory.generateActorId("test-actor-");
74 InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, id));
75 InMemoryJournal.addEntry(id, 2, new SimpleReplicatedLogEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
76 InMemoryJournal.addEntry(id, 3, new ApplyJournalEntries(0));
78 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
79 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
81 RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() {
83 public void createSnapshot(ActorRef actorRef, java.util.Optional<OutputStream> installSnapshotStream) {
84 actorRef.tell(new CaptureSnapshotReply(ByteState.empty(), installSnapshotStream), actorRef);
88 public void applySnapshot(Snapshot.State snapshotState) {
92 public State deserializeSnapshot(ByteSource snapshotBytes) {
93 throw new UnsupportedOperationException();
97 TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(id)
98 .config(config).snapshotCohort(snapshotCohort).persistent(Optional.of(true)).props()
99 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
100 MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
102 mockRaftActor.waitForRecoveryComplete();
104 Uninterruptibles.sleepUninterruptibly(750, TimeUnit.MILLISECONDS);
106 List<Snapshot> snapshots = InMemorySnapshotStore.getSnapshots(id, Snapshot.class);
107 assertEquals("Snapshots", 0, snapshots.size());
109 TEST_LOG.info("testNoSnapshotAfterStartupWithNoMigratedMessages ending");
113 public void testSnapshotAfterStartupWithMigratedSnapshot() throws Exception {
114 TEST_LOG.info("testSnapshotAfterStartupWithMigratedSnapshot starting");
116 String persistenceId = factory.generateActorId("test-actor-");
118 List<Object> snapshotData = Arrays.asList(new MockPayload("1"));
119 final MockSnapshotState snapshotState = new MockSnapshotState(snapshotData);
121 org.opendaylight.controller.cluster.raft.Snapshot legacy = org.opendaylight.controller.cluster.raft.Snapshot
122 .create(SerializationUtils.serialize((Serializable) snapshotData),
123 Arrays.asList(new SimpleReplicatedLogEntry(6, 2, new MockPayload("payload"))),
124 6, 2, 5, 1, 3, "member-1", new ServerConfigurationPayload(Arrays.asList(
125 new ServerInfo(persistenceId, true), new ServerInfo("2", false))));
126 InMemorySnapshotStore.addSnapshot(persistenceId, legacy);
128 doTestSnapshotAfterStartupWithMigratedMessage(persistenceId, true, snapshot -> {
129 assertEquals("getLastIndex", legacy.getLastIndex(), snapshot.getLastIndex());
130 assertEquals("getLastTerm", legacy.getLastTerm(), snapshot.getLastTerm());
131 assertEquals("getLastAppliedIndex", legacy.getLastAppliedIndex(), snapshot.getLastAppliedIndex());
132 assertEquals("getLastAppliedTerm", legacy.getLastAppliedTerm(), snapshot.getLastAppliedTerm());
133 assertEquals("getState", snapshotState, snapshot.getState());
134 assertEquals("Unapplied entries size", legacy.getUnAppliedEntries().size(),
135 snapshot.getUnAppliedEntries().size());
136 assertEquals("Unapplied entry term", legacy.getUnAppliedEntries().get(0).getTerm(),
137 snapshot.getUnAppliedEntries().get(0).getTerm());
138 assertEquals("Unapplied entry index", legacy.getUnAppliedEntries().get(0).getIndex(),
139 snapshot.getUnAppliedEntries().get(0).getIndex());
140 assertEquals("Unapplied entry data", legacy.getUnAppliedEntries().get(0).getData(),
141 snapshot.getUnAppliedEntries().get(0).getData());
142 assertEquals("getElectionVotedFor", legacy.getElectionVotedFor(), snapshot.getElectionVotedFor());
143 assertEquals("getElectionTerm", legacy.getElectionTerm(), snapshot.getElectionTerm());
144 assertEquals("getServerConfiguration", Sets.newHashSet(legacy.getServerConfiguration().getServerConfig()),
145 Sets.newHashSet(snapshot.getServerConfiguration().getServerConfig()));
148 TEST_LOG.info("testSnapshotAfterStartupWithMigratedSnapshot ending");
151 @SuppressWarnings("checkstyle:IllegalCatch")
152 private TestActorRef<MockRaftActor> doTestSnapshotAfterStartupWithMigratedMessage(String id, boolean persistent,
153 Consumer<Snapshot> snapshotVerifier, final State snapshotState) {
154 InMemorySnapshotStore.addSnapshotSavedLatch(id);
155 InMemoryJournal.addDeleteMessagesCompleteLatch(id);
156 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
157 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
159 RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() {
161 public void createSnapshot(ActorRef actorRef, java.util.Optional<OutputStream> installSnapshotStream) {
162 actorRef.tell(new CaptureSnapshotReply(snapshotState, installSnapshotStream), actorRef);
166 public void applySnapshot(State newState) {
170 public State deserializeSnapshot(ByteSource snapshotBytes) {
171 throw new UnsupportedOperationException();
175 TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(id)
176 .config(config).snapshotCohort(snapshotCohort).persistent(Optional.of(persistent))
177 .peerAddresses(ImmutableMap.of("peer", "")).props()
178 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
179 MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
181 mockRaftActor.waitForRecoveryComplete();
183 Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
184 snapshotVerifier.accept(snapshot);
186 InMemoryJournal.waitForDeleteMessagesComplete(id);
188 assertEquals("InMemoryJournal size", 0, InMemoryJournal.get(id).size());