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=d02819a4925a12fd00de323e62bdb63c64b799b6;hpb=35235f427f3a056f85fe83ddd1133e67540328f7;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 d02819a492..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
@@ -7,6 +7,7 @@
  */
 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;
@@ -14,7 +15,6 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyObject;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.doReturn;
@@ -49,6 +49,8 @@ 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;
@@ -413,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
@@ -585,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());
 
@@ -675,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());
 
@@ -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();
@@ -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<MockRaftActor> 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 <T> void persistAsync(final T entry, final Procedure<T> 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();
+    }
 }