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%2FRaftActorTest.java;h=941deb5843e2749a08232ba8b1cf5a55921419fb;hb=f1c3050779d7770ef6a12a67a1870765c3dfd9eb;hp=a2382379f205db9c7f6228b7b1d3886daeba8b80;hpb=023402c7c80372260b6c5c82f120093a73806717;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java index a2382379f2..941deb5843 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java @@ -8,9 +8,12 @@ package org.opendaylight.controller.cluster.raft; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyObject; @@ -18,14 +21,17 @@ import static org.mockito.Matchers.eq; import static org.mockito.Matchers.same; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; import akka.actor.ActorRef; import akka.actor.PoisonPill; import akka.actor.Props; +import akka.actor.Status.Failure; import akka.actor.Terminated; import akka.dispatch.Dispatchers; import akka.japi.Procedure; -import akka.persistence.RecoveryCompleted; import akka.persistence.SaveSnapshotFailure; import akka.persistence.SaveSnapshotSuccess; import akka.persistence.SnapshotMetadata; @@ -45,9 +51,12 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import org.apache.commons.lang3.SerializationUtils; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.NonPersistentDataProvider; import org.opendaylight.controller.cluster.PersistentDataProvider; @@ -66,13 +75,17 @@ import org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm import org.opendaylight.controller.cluster.raft.behaviors.Follower; import org.opendaylight.controller.cluster.raft.behaviors.Leader; import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior; +import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot; +import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshotReply; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; +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.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; public class RaftActorTest extends AbstractActorTest { @@ -104,6 +117,7 @@ public class RaftActorTest extends AbstractActorTest { kit.waitUntilLeader(); } + @Test public void testRaftActorRecoveryWithPersistenceEnabled() throws Exception { TEST_LOG.info("testRaftActorRecoveryWithPersistenceEnabled starting"); @@ -337,34 +351,40 @@ public class RaftActorTest extends AbstractActorTest { mockRaftActor.waitForRecoveryComplete(); ApplySnapshot applySnapshot = new ApplySnapshot(mock(Snapshot.class)); - doReturn(true).when(mockSupport).handleSnapshotMessage(same(applySnapshot)); + doReturn(true).when(mockSupport).handleSnapshotMessage(same(applySnapshot), any(ActorRef.class)); mockRaftActor.handleCommand(applySnapshot); CaptureSnapshot captureSnapshot = new CaptureSnapshot(1, 1, 1, 1, 0, 1, null); - doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshot)); + doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshot), any(ActorRef.class)); mockRaftActor.handleCommand(captureSnapshot); CaptureSnapshotReply captureSnapshotReply = new CaptureSnapshotReply(new byte[0]); - doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshotReply)); + doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshotReply), any(ActorRef.class)); mockRaftActor.handleCommand(captureSnapshotReply); SaveSnapshotSuccess saveSnapshotSuccess = new SaveSnapshotSuccess(mock(SnapshotMetadata.class)); - doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess)); + doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess), any(ActorRef.class)); mockRaftActor.handleCommand(saveSnapshotSuccess); SaveSnapshotFailure saveSnapshotFailure = new SaveSnapshotFailure(mock(SnapshotMetadata.class), new Throwable()); - doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure)); + doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure), any(ActorRef.class)); mockRaftActor.handleCommand(saveSnapshotFailure); - doReturn(true).when(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT)); + doReturn(true).when(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT), + any(ActorRef.class)); mockRaftActor.handleCommand(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT); - verify(mockSupport).handleSnapshotMessage(same(applySnapshot)); - verify(mockSupport).handleSnapshotMessage(same(captureSnapshot)); - verify(mockSupport).handleSnapshotMessage(same(captureSnapshotReply)); - verify(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess)); - verify(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure)); - verify(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT)); + doReturn(true).when(mockSupport).handleSnapshotMessage(same(GetSnapshot.INSTANCE), any(ActorRef.class)); + mockRaftActor.handleCommand(GetSnapshot.INSTANCE); + + verify(mockSupport).handleSnapshotMessage(same(applySnapshot), any(ActorRef.class)); + verify(mockSupport).handleSnapshotMessage(same(captureSnapshot), any(ActorRef.class)); + verify(mockSupport).handleSnapshotMessage(same(captureSnapshotReply), any(ActorRef.class)); + verify(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess), any(ActorRef.class)); + verify(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure), any(ActorRef.class)); + verify(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT), + any(ActorRef.class)); + verify(mockSupport).handleSnapshotMessage(same(GetSnapshot.INSTANCE), any(ActorRef.class)); } @Test @@ -942,6 +962,40 @@ public class RaftActorTest extends AbstractActorTest { }}; } + @Test + public void testRaftActorOnRecoverySnapshot() throws Exception { + TEST_LOG.info("testRaftActorOnRecoverySnapshot"); + + new JavaTestKit(getSystem()) {{ + String persistenceId = factory.generateActorId("follower-"); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + + // Set the heartbeat interval high to essentially disable election otherwise the test + // may fail if the actor is switched to Leader + config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + ImmutableMap peerAddresses = ImmutableMap.builder().put("member1", "address").build(); + + // Create mock ReplicatedLogEntry + ReplicatedLogEntry replLogEntry = new MockRaftActorContext.MockReplicatedLogEntry(1,1, + new MockRaftActorContext.MockPayload("F", 1)); + + InMemoryJournal.addEntry(persistenceId, 1, replLogEntry); + + TestActorRef ref = factory.createTestActor( + MockRaftActor.props(persistenceId, peerAddresses, Optional.of(config))); + + MockRaftActor mockRaftActor = ref.underlyingActor(); + + mockRaftActor.waitForRecoveryComplete(); + + mockRaftActor.waitForInitializeBehaviorComplete(); + + verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(any(ActorRef.class)); + }}; + } + @Test public void testSwitchBehavior(){ String persistenceId = factory.generateActorId("leader-"); @@ -961,7 +1015,7 @@ public class RaftActorTest extends AbstractActorTest { MockRaftActor leaderActor = mockActorRef.underlyingActor(); - leaderActor.handleRecover(RecoveryCompleted.getInstance()); + leaderActor.waitForRecoveryComplete(); leaderActor.handleCommand(new SwitchBehavior(RaftState.Follower, 100)); @@ -982,8 +1036,6 @@ public class RaftActorTest extends AbstractActorTest { assertEquals(110, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm()); assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state()); - - } public static ByteString fromObject(Object snapshot) throws Exception { @@ -1006,4 +1058,141 @@ public class RaftActorTest extends AbstractActorTest { } } + @Test + public void testUpdateConfigParam() throws Exception { + DefaultConfigParamsImpl emptyConfig = new DefaultConfigParamsImpl(); + String persistenceId = factory.generateActorId("follower-"); + ImmutableMap peerAddresses = + ImmutableMap.builder().put("member1", "address").build(); + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); + + TestActorRef actorRef = factory.createTestActor( + MockRaftActor.props(persistenceId, peerAddresses, + Optional.of(emptyConfig), dataPersistenceProvider), persistenceId); + MockRaftActor mockRaftActor = actorRef.underlyingActor(); + mockRaftActor.waitForInitializeBehaviorComplete(); + + RaftActorBehavior behavior = mockRaftActor.getCurrentBehavior(); + mockRaftActor.updateConfigParams(emptyConfig); + assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior()); + assertEquals("Behavior State", RaftState.Follower, + mockRaftActor.getCurrentBehavior().state()); + + DefaultConfigParamsImpl disableConfig = new DefaultConfigParamsImpl(); + disableConfig.setCustomRaftPolicyImplementationClass( + "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy"); + mockRaftActor.updateConfigParams(disableConfig); + assertNotSame("Different Behavior", behavior, mockRaftActor.getCurrentBehavior()); + assertEquals("Behavior State", RaftState.Follower, + mockRaftActor.getCurrentBehavior().state()); + + behavior = mockRaftActor.getCurrentBehavior(); + mockRaftActor.updateConfigParams(disableConfig); + assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior()); + assertEquals("Behavior State", RaftState.Follower, + mockRaftActor.getCurrentBehavior().state()); + + DefaultConfigParamsImpl defaultConfig = new DefaultConfigParamsImpl(); + defaultConfig.setCustomRaftPolicyImplementationClass( + "org.opendaylight.controller.cluster.raft.policy.DefaultRaftPolicy"); + mockRaftActor.updateConfigParams(defaultConfig); + assertNotSame("Different Behavior", behavior, mockRaftActor.getCurrentBehavior()); + assertEquals("Behavior State", RaftState.Follower, + mockRaftActor.getCurrentBehavior().state()); + + behavior = mockRaftActor.getCurrentBehavior(); + mockRaftActor.updateConfigParams(defaultConfig); + assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior()); + assertEquals("Behavior State", RaftState.Follower, + mockRaftActor.getCurrentBehavior().state()); + } + + @Test + public void testGetSnapshot() throws Exception { + TEST_LOG.info("testGetSnapshot starting"); + + JavaTestKit kit = new JavaTestKit(getSystem()); + + String persistenceId = factory.generateActorId("test-actor-"); + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName()); + + long term = 3; + long seqN = 1; + InMemoryJournal.addEntry(persistenceId, seqN++, new UpdateElectionTerm(term, "member-1")); + InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 0, + new MockRaftActorContext.MockPayload("A"))); + InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 1, + new MockRaftActorContext.MockPayload("B"))); + InMemoryJournal.addEntry(persistenceId, seqN++, new ApplyJournalEntries(1)); + InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 2, + new MockRaftActorContext.MockPayload("C"))); + + TestActorRef raftActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + ImmutableMap.builder().put("member1", "address").build(), Optional.of(config)). + withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId); + MockRaftActor mockRaftActor = raftActorRef.underlyingActor(); + + mockRaftActor.waitForRecoveryComplete(); + + // Wait for snapshot after recovery + verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(any(ActorRef.class)); + + mockRaftActor.snapshotCohortDelegate = mock(RaftActorSnapshotCohort.class); + + raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef()); + + ArgumentCaptor replyActor = ArgumentCaptor.forClass(ActorRef.class); + verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(replyActor.capture()); + + byte[] stateSnapshot = new byte[]{1,2,3}; + replyActor.getValue().tell(new CaptureSnapshotReply(stateSnapshot), ActorRef.noSender()); + + GetSnapshotReply reply = kit.expectMsgClass(GetSnapshotReply.class); + + assertEquals("getId", persistenceId, reply.getId()); + Snapshot replySnapshot = SerializationUtils.deserialize(reply.getSnapshot()); + assertEquals("getElectionTerm", term, replySnapshot.getElectionTerm()); + assertEquals("getElectionVotedFor", "member-1", replySnapshot.getElectionVotedFor()); + assertEquals("getLastAppliedIndex", 1L, replySnapshot.getLastAppliedIndex()); + assertEquals("getLastAppliedTerm", term, replySnapshot.getLastAppliedTerm()); + assertEquals("getLastIndex", 2L, replySnapshot.getLastIndex()); + assertEquals("getLastTerm", term, replySnapshot.getLastTerm()); + assertArrayEquals("getState", stateSnapshot, replySnapshot.getState()); + assertEquals("getUnAppliedEntries size", 1, replySnapshot.getUnAppliedEntries().size()); + assertEquals("UnApplied entry index ", 2L, replySnapshot.getUnAppliedEntries().get(0).getIndex()); + + // Test with timeout + + mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout(Duration.create(200, TimeUnit.MILLISECONDS)); + reset(mockRaftActor.snapshotCohortDelegate); + + raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef()); + Failure failure = kit.expectMsgClass(akka.actor.Status.Failure.class); + assertEquals("Failure cause type", TimeoutException.class, failure.cause().getClass()); + + mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout(Duration.create(30, TimeUnit.SECONDS)); + + // Test with persistence disabled. + + mockRaftActor.setPersistence(false); + reset(mockRaftActor.snapshotCohortDelegate); + + raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef()); + reply = kit.expectMsgClass(GetSnapshotReply.class); + verify(mockRaftActor.snapshotCohortDelegate, never()).createSnapshot(any(ActorRef.class)); + + assertEquals("getId", persistenceId, reply.getId()); + replySnapshot = SerializationUtils.deserialize(reply.getSnapshot()); + assertEquals("getElectionTerm", term, replySnapshot.getElectionTerm()); + assertEquals("getElectionVotedFor", "member-1", replySnapshot.getElectionVotedFor()); + assertEquals("getLastAppliedIndex", -1L, replySnapshot.getLastAppliedIndex()); + assertEquals("getLastAppliedTerm", -1L, replySnapshot.getLastAppliedTerm()); + assertEquals("getLastIndex", -1L, replySnapshot.getLastIndex()); + assertEquals("getLastTerm", -1L, replySnapshot.getLastTerm()); + assertEquals("getState length", 0, replySnapshot.getState().length); + assertEquals("getUnAppliedEntries size", 0, replySnapshot.getUnAppliedEntries().size()); + + TEST_LOG.info("testGetSnapshot ending"); + } }