X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FMigratedMessagesTest.java;h=5a1ac54c7cab6f8a477bce09d670e80463269c84;hb=f33beecf2a10955a9219757529ba3017079816cc;hp=2f1a39174b85532f7d083b0bf35514cb2996080b;hpb=92cbb07ef81943b0740ba7c8915001ac6785f560;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MigratedMessagesTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MigratedMessagesTest.java index 2f1a39174b..5a1ac54c7c 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MigratedMessagesTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MigratedMessagesTest.java @@ -8,10 +8,15 @@ package org.opendaylight.controller.cluster.raft; import static org.junit.Assert.assertEquals; + +import akka.actor.ActorRef; +import akka.dispatch.Dispatchers; +import akka.testkit.TestActorRef; import com.google.common.base.Optional; +import com.google.common.collect.ImmutableMap; +import com.google.common.io.ByteSource; import com.google.common.util.concurrent.Uninterruptibles; -import java.util.Arrays; -import java.util.HashSet; +import java.io.OutputStream; import java.util.List; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; @@ -19,18 +24,17 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; -import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload; -import org.opendaylight.controller.cluster.raft.persisted.ServerInfo; +import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries; +import org.opendaylight.controller.cluster.raft.persisted.ByteState; +import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.persisted.Snapshot; +import org.opendaylight.controller.cluster.raft.persisted.Snapshot.State; import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm; import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import akka.actor.ActorRef; -import akka.dispatch.Dispatchers; -import akka.testkit.JavaTestKit; -import akka.testkit.TestActorRef; /** * Unit tests for migrated messages on recovery. @@ -43,129 +47,48 @@ public class MigratedMessagesTest extends AbstractActorTest { private TestActorFactory factory; @Before - public void setUp(){ + public void setUp() { factory = new TestActorFactory(getSystem()); } @After - public void tearDown() throws Exception { + public void tearDown() { factory.close(); InMemoryJournal.clear(); InMemorySnapshotStore.clear(); } - @Test - public void testSnapshotAfterStartupWithMigratedServerConfigPayloadAndPersistenceEnabled() { - TEST_LOG.info("testSnapshotAfterStartupWithMigratedServerConfigPayloadAndPersistenceEnabled starting"); - doTestSnapshotAfterStartupWithMigratedServerConfigPayload(true); - TEST_LOG.info("testSnapshotAfterStartupWithMigratedServerConfigPayloadAndPersistenceEnabled ending"); - } - - @Test - public void testSnapshotAfterStartupWithMigratedServerConfigPayloadAndPersistenceDisabled() { - TEST_LOG.info("testSnapshotAfterStartupWithMigratedServerConfigPayloadAndPersistenceDisabled starting"); - - TestActorRef actor = doTestSnapshotAfterStartupWithMigratedServerConfigPayload(false); - MockRaftActor mockRaftActor = actor.underlyingActor(); - String id = mockRaftActor.persistenceId(); - ConfigParams config = mockRaftActor.getRaftActorContext().getConfigParams(); - - factory.killActor(actor, new JavaTestKit(getSystem())); - - actor = factory.createTestActor(MockRaftActor.builder().id(id).config(config).persistent(Optional.of(false)).props(). - withDispatcher(Dispatchers.DefaultDispatcherId()), id); - mockRaftActor = actor.underlyingActor(); - mockRaftActor.waitForRecoveryComplete(); - - assertEquals("electionTerm", 1, - mockRaftActor.getRaftActorContext().getTermInformation().getCurrentTerm()); - assertEquals("votedFor", id, - mockRaftActor.getRaftActorContext().getTermInformation().getVotedFor()); - - TEST_LOG.info("testSnapshotAfterStartupWithMigratedServerConfigPayloadAndPersistenceDisabled ending"); - } - - @Test - public void testSnapshotAfterStartupWithMigratedUpdateElectionTermAndPersistenceEnabled() { - TEST_LOG.info("testSnapshotAfterStartupWithMigratedUpdateElectionTermAndPersistenceEnabled starting"); - - String persistenceId = factory.generateActorId("test-actor-"); - - org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm updateElectionTerm = - new org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm(5, persistenceId); - - InMemoryJournal.addEntry(persistenceId, 1, updateElectionTerm); - - doTestSnapshotAfterStartupWithMigratedMessage(persistenceId, true, snapshot -> { - assertEquals("getElectionVotedFor", persistenceId, snapshot.getElectionVotedFor()); - assertEquals("getElectionTerm", 5, snapshot.getElectionTerm()); - }); - - TEST_LOG.info("testSnapshotAfterStartupWithMigratedUpdateElectionTermAndPersistenceEnabled ending"); - } - - @Test - public void testSnapshotAfterStartupWithMigratedUpdateElectionTermAndPersistenceDisabled() { - TEST_LOG.info("testSnapshotAfterStartupWithMigratedUpdateElectionTermAndPersistenceDisabled starting"); - - String persistenceId = factory.generateActorId("test-actor-"); - - org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm updateElectionTerm = - new org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm(5, persistenceId); - - InMemoryJournal.addEntry(persistenceId, 1, updateElectionTerm); - - doTestSnapshotAfterStartupWithMigratedMessage(persistenceId, false, snapshot -> { - assertEquals("getElectionVotedFor", persistenceId, snapshot.getElectionVotedFor()); - assertEquals("getElectionTerm", 5, snapshot.getElectionTerm()); - }); - - TEST_LOG.info("testSnapshotAfterStartupWithMigratedUpdateElectionTermAndPersistenceDisabled ending"); - } - - @Test - public void testSnapshotAfterStartupWithMigratedApplyJournalEntries() { - TEST_LOG.info("testSnapshotAfterStartupWithMigratedApplyJournalEntries starting"); - - String persistenceId = factory.generateActorId("test-actor-"); - - InMemoryJournal.addEntry(persistenceId, 1, new UpdateElectionTerm(1, persistenceId)); - InMemoryJournal.addEntry(persistenceId, 2, new ReplicatedLogImplEntry(0, 1, - new MockRaftActorContext.MockPayload("A"))); - InMemoryJournal.addEntry(persistenceId, 3, - new org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries(0)); - - - doTestSnapshotAfterStartupWithMigratedMessage(persistenceId, true, snapshot -> { - assertEquals("getLastAppliedIndex", 0, snapshot.getLastAppliedIndex()); - assertEquals("getLastAppliedTerm", 1, snapshot.getLastAppliedTerm()); - assertEquals("getLastIndex", 0, snapshot.getLastIndex()); - assertEquals("getLastTerm", 1, snapshot.getLastTerm()); - }); - - TEST_LOG.info("testSnapshotAfterStartupWithMigratedApplyJournalEntries ending"); - } - @Test public void testNoSnapshotAfterStartupWithNoMigratedMessages() { + TEST_LOG.info("testNoSnapshotAfterStartupWithNoMigratedMessages starting"); String id = factory.generateActorId("test-actor-"); + + InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, id)); + InMemoryJournal.addEntry(id, 2, new SimpleReplicatedLogEntry(0, 1, new MockRaftActorContext.MockPayload("A"))); + InMemoryJournal.addEntry(id, 3, new ApplyJournalEntries(0)); + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName()); RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() { @Override - public void createSnapshot(ActorRef actorRef) { - actorRef.tell(new CaptureSnapshotReply(new byte[0]), actorRef); + public void createSnapshot(ActorRef actorRef, java.util.Optional installSnapshotStream) { + actorRef.tell(new CaptureSnapshotReply(ByteState.empty(), installSnapshotStream), actorRef); + } + + @Override + public void applySnapshot(Snapshot.State snapshotState) { } @Override - public void applySnapshot(byte[] snapshotBytes) { + public State deserializeSnapshot(ByteSource snapshotBytes) { + throw new UnsupportedOperationException(); } }; - TestActorRef raftActorRef = factory.createTestActor(MockRaftActor.builder().id(id). - config(config).snapshotCohort(snapshotCohort).persistent(Optional.of(true)).props(). - withDispatcher(Dispatchers.DefaultDispatcherId()), id); + TestActorRef raftActorRef = factory.createTestActor(MockRaftActor.builder().id(id) + .config(config).snapshotCohort(snapshotCohort).persistent(Optional.of(true)).props() + .withDispatcher(Dispatchers.DefaultDispatcherId()), id); MockRaftActor mockRaftActor = raftActorRef.underlyingActor(); mockRaftActor.waitForRecoveryComplete(); @@ -174,38 +97,13 @@ public class MigratedMessagesTest extends AbstractActorTest { List snapshots = InMemorySnapshotStore.getSnapshots(id, Snapshot.class); assertEquals("Snapshots", 0, snapshots.size()); - } - - private TestActorRef doTestSnapshotAfterStartupWithMigratedServerConfigPayload(boolean persistent) { - String persistenceId = factory.generateActorId("test-actor-"); - - org.opendaylight.controller.cluster.raft.ServerConfigurationPayload persistedServerConfig = - new org.opendaylight.controller.cluster.raft.ServerConfigurationPayload(Arrays.asList( - new org.opendaylight.controller.cluster.raft.ServerConfigurationPayload.ServerInfo( - persistenceId, true), - new org.opendaylight.controller.cluster.raft.ServerConfigurationPayload.ServerInfo( - "downNode", true))); - ServerConfigurationPayload expectedServerConfig = new ServerConfigurationPayload(Arrays.asList( - new ServerInfo(persistenceId, true), new ServerInfo("downNode", true))); - - InMemoryJournal.addEntry(persistenceId, 1, new UpdateElectionTerm(1, persistenceId)); - InMemoryJournal.addEntry(persistenceId, 3, new ReplicatedLogImplEntry(0, 1, persistedServerConfig)); - - TestActorRef actor = doTestSnapshotAfterStartupWithMigratedMessage(persistenceId, - persistent, snapshot -> { - assertEquals("getElectionVotedFor", persistenceId, snapshot.getElectionVotedFor()); - assertEquals("getElectionTerm", 1, snapshot.getElectionTerm()); - assertEquals("getServerConfiguration", new HashSet<>(expectedServerConfig.getServerConfig()), - new HashSet<>(snapshot.getServerConfiguration().getServerConfig())); - }); - - return actor; + TEST_LOG.info("testNoSnapshotAfterStartupWithNoMigratedMessages ending"); } - + @SuppressWarnings("checkstyle:IllegalCatch") private TestActorRef doTestSnapshotAfterStartupWithMigratedMessage(String id, boolean persistent, - Consumer snapshotVerifier) { + Consumer snapshotVerifier, final State snapshotState) { InMemorySnapshotStore.addSnapshotSavedLatch(id); InMemoryJournal.addDeleteMessagesCompleteLatch(id); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); @@ -213,18 +111,24 @@ public class MigratedMessagesTest extends AbstractActorTest { RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() { @Override - public void createSnapshot(ActorRef actorRef) { - actorRef.tell(new CaptureSnapshotReply(new byte[0]), actorRef); + public void createSnapshot(ActorRef actorRef, java.util.Optional installSnapshotStream) { + actorRef.tell(new CaptureSnapshotReply(snapshotState, installSnapshotStream), actorRef); + } + + @Override + public void applySnapshot(State newState) { } @Override - public void applySnapshot(byte[] snapshotBytes) { + public State deserializeSnapshot(ByteSource snapshotBytes) { + throw new UnsupportedOperationException(); } }; - TestActorRef raftActorRef = factory.createTestActor(MockRaftActor.builder().id(id). - config(config).snapshotCohort(snapshotCohort).persistent(Optional.of(persistent)).props(). - withDispatcher(Dispatchers.DefaultDispatcherId()), id); + TestActorRef raftActorRef = factory.createTestActor(MockRaftActor.builder().id(id) + .config(config).snapshotCohort(snapshotCohort).persistent(Optional.of(persistent)) + .peerAddresses(ImmutableMap.of("peer", "")).props() + .withDispatcher(Dispatchers.DefaultDispatcherId()), id); MockRaftActor mockRaftActor = raftActorRef.underlyingActor(); mockRaftActor.waitForRecoveryComplete();