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=ce7dd98fc44276d7c5ea6db959dd838f002b9fb6;hb=5b685df28d9ea19494ffc5a998e7059ab15a118e;hp=56089eca4d976a774ed46a7290434283fd61511d;hpb=e9fc7e7ed2b13d274518d6a872ab67749ef4507a;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 56089eca4d..ce7dd98fc4 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 @@ -5,19 +5,18 @@ * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ - package org.opendaylight.controller.cluster.raft; +import static org.awaitility.Awaitility.await; 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; -import static org.mockito.Matchers.eq; -import static org.mockito.Matchers.same; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.same; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -35,20 +34,23 @@ import akka.persistence.SaveSnapshotFailure; import akka.persistence.SaveSnapshotSuccess; import akka.persistence.SnapshotMetadata; import akka.persistence.SnapshotOffer; +import akka.protobuf.ByteString; import akka.testkit.TestActorRef; import akka.testkit.javadsl.TestKit; import com.google.common.base.Optional; import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.Uninterruptibles; -import com.google.protobuf.ByteString; import java.io.ByteArrayOutputStream; import java.io.ObjectOutputStream; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.junit.After; @@ -60,6 +62,8 @@ import org.opendaylight.controller.cluster.NonPersistentDataProvider; import org.opendaylight.controller.cluster.PersistentDataProvider; import org.opendaylight.controller.cluster.notifications.LeaderStateChanged; import org.opendaylight.controller.cluster.notifications.RoleChanged; +import org.opendaylight.controller.cluster.raft.AbstractRaftActorIntegrationTest.TestPersist; +import org.opendaylight.controller.cluster.raft.AbstractRaftActorIntegrationTest.TestRaftActor; import org.opendaylight.controller.cluster.raft.MockRaftActor.MockSnapshotState; import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; @@ -91,7 +95,6 @@ import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; import org.opendaylight.yangtools.concepts.Identifier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; public class RaftActorTest extends AbstractActorTest { @@ -106,7 +109,7 @@ public class RaftActorTest extends AbstractActorTest { } @After - public void tearDown() throws Exception { + public void tearDown() { factory.close(); InMemoryJournal.clear(); InMemorySnapshotStore.clear(); @@ -125,7 +128,7 @@ public class RaftActorTest extends AbstractActorTest { @Test - public void testRaftActorRecoveryWithPersistenceEnabled() throws Exception { + public void testRaftActorRecoveryWithPersistenceEnabled() { TEST_LOG.info("testRaftActorRecoveryWithPersistenceEnabled starting"); TestKit kit = new TestKit(getSystem()); @@ -183,7 +186,7 @@ public class RaftActorTest extends AbstractActorTest { // kill the actor followerActor.tell(PoisonPill.getInstance(), null); - kit.expectMsgClass(kit.duration("5 seconds"), Terminated.class); + kit.expectMsgClass(Duration.ofSeconds(5), Terminated.class); kit.unwatch(followerActor); @@ -212,7 +215,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testRaftActorRecoveryWithPersistenceDisabled() throws Exception { + public void testRaftActorRecoveryWithPersistenceDisabled() { String persistenceId = factory.generateActorId("follower-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); @@ -233,7 +236,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testUpdateElectionTermPersistedWithPersistenceDisabled() throws Exception { + public void testUpdateElectionTermPersistedWithPersistenceDisabled() { final TestKit kit = new TestKit(getSystem()); String persistenceId = factory.generateActorId("follower-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); @@ -393,7 +396,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testApplyState() throws Exception { + public void testApplyState() { String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); @@ -414,7 +417,7 @@ public class RaftActorTest extends AbstractActorTest { final Identifier id = new MockIdentifier("apply-state"); mockRaftActor.getRaftActorContext().getApplyStateConsumer().accept(new ApplyState(mockActorRef, id, entry)); - verify(mockRaftActor.actorDelegate).applyState(eq(mockActorRef), eq(id), anyObject()); + verify(mockRaftActor.actorDelegate).applyState(eq(mockActorRef), eq(id), any()); } @Test @@ -586,7 +589,7 @@ public class RaftActorTest extends AbstractActorTest { leaderActor.getRaftActorContext().getSnapshotManager().capture( new SimpleReplicatedLogEntry(6, 1, new MockRaftActorContext.MockPayload("x")), 4); - verify(leaderActor.snapshotCohortDelegate).createSnapshot(anyObject(), anyObject()); + verify(leaderActor.snapshotCohortDelegate).createSnapshot(any(), any()); assertEquals(8, leaderActor.getReplicatedLog().size()); @@ -676,7 +679,7 @@ public class RaftActorTest extends AbstractActorTest { followerActor.getRaftActorContext().getSnapshotManager().capture( new SimpleReplicatedLogEntry(5, 1, new MockRaftActorContext.MockPayload("D")), 4); - verify(followerActor.snapshotCohortDelegate).createSnapshot(anyObject(), anyObject()); + verify(followerActor.snapshotCohortDelegate).createSnapshot(any(), any()); assertEquals(6, followerActor.getReplicatedLog().size()); @@ -958,7 +961,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testUpdateConfigParam() throws Exception { + public void testUpdateConfigParam() { DefaultConfigParamsImpl emptyConfig = new DefaultConfigParamsImpl(); String persistenceId = factory.generateActorId("follower-"); ImmutableMap peerAddresses = @@ -1006,7 +1009,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testGetSnapshot() throws Exception { + public void testGetSnapshot() { TEST_LOG.info("testGetSnapshot starting"); final TestKit kit = new TestKit(getSystem()); @@ -1062,14 +1065,15 @@ public class RaftActorTest extends AbstractActorTest { // Test with timeout mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout( - Duration.create(200, TimeUnit.MILLISECONDS)); + FiniteDuration.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)); + mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout( + FiniteDuration.create(30, TimeUnit.SECONDS)); // Test with persistence disabled. @@ -1078,7 +1082,7 @@ public class RaftActorTest extends AbstractActorTest { raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef()); reply = kit.expectMsgClass(GetSnapshotReply.class); - verify(mockRaftActor.snapshotCohortDelegate, never()).createSnapshot(anyObject(), anyObject()); + verify(mockRaftActor.snapshotCohortDelegate, never()).createSnapshot(any(), any()); assertEquals("getId", persistenceId, reply.getId()); replySnapshot = reply.getSnapshot(); @@ -1095,7 +1099,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testRestoreFromSnapshot() throws Exception { + public void testRestoreFromSnapshot() { TEST_LOG.info("testRestoreFromSnapshot starting"); String persistenceId = factory.generateActorId("test-actor-"); @@ -1208,7 +1212,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testNonVotingOnRecovery() throws Exception { + public void testNonVotingOnRecovery() { TEST_LOG.info("testNonVotingOnRecovery starting"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); @@ -1234,7 +1238,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testLeaderTransitioning() throws Exception { + public void testLeaderTransitioning() { TEST_LOG.info("testLeaderTransitioning starting"); ActorRef notifierActor = factory.createActor(MessageCollectorActor.props()); @@ -1351,4 +1355,68 @@ public class RaftActorTest extends AbstractActorTest { AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("AppendEntries size", 3, appendEntries.getEntries().size()); } + + @Test + @SuppressWarnings("checkstyle:illegalcatch") + public void testApplyStateRace() throws Exception { + final String leaderId = factory.generateActorId("leader-"); + final String followerId = factory.generateActorId("follower-"); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS)); + config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName()); + + ActorRef mockFollowerActorRef = factory.createActor(MessageCollectorActor.props()); + + TestRaftActor.Builder builder = TestRaftActor.newBuilder() + .id(leaderId) + .peerAddresses(ImmutableMap.of(followerId, + mockFollowerActorRef.path().toString())) + .config(config) + .collectorActor(factory.createActor( + MessageCollectorActor.props(), factory.generateActorId(leaderId + "-collector"))); + + TestActorRef leaderActorRef = factory.createTestActor( + builder.props(), leaderId); + MockRaftActor leaderActor = leaderActorRef.underlyingActor(); + leaderActor.waitForInitializeBehaviorComplete(); + + leaderActor.getRaftActorContext().getTermInformation().update(1, leaderId); + Leader leader = new Leader(leaderActor.getRaftActorContext()); + leaderActor.setCurrentBehavior(leader); + + final ExecutorService executorService = Executors.newSingleThreadExecutor(); + + leaderActor.setPersistence(new PersistentDataProvider(leaderActor) { + @Override + public void persistAsync(final T entry, final Procedure procedure) { + // needs to be executed from another thread to simulate the persistence actor calling this callback + executorService.submit(() -> { + try { + procedure.apply(entry); + } catch (Exception e) { + TEST_LOG.info("Fail during async persist callback", e); + } + }, "persistence-callback"); + } + }); + + leader.getFollower(followerId).setNextIndex(0); + leader.getFollower(followerId).setMatchIndex(-1); + + // hitting this is flimsy so run multiple times to improve the chance of things + // blowing up while breaking actor containment + final TestPersist message = + new TestPersist(leaderActorRef, new MockIdentifier("1"), new MockPayload("1")); + for (int i = 0; i < 100; i++) { + leaderActorRef.tell(message, null); + + AppendEntriesReply reply = + new AppendEntriesReply(followerId, 1, true, i, 1, (short) 5); + leaderActorRef.tell(reply, mockFollowerActorRef); + } + + await("Persistence callback.").atMost(5, TimeUnit.SECONDS).until(() -> leaderActor.getState().size() == 100); + executorService.shutdown(); + } }