*/
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 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 java.io.ByteArrayOutputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.junit.After;
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;
// Wait for akka's recovery to complete so it doesn't interfere.
mockRaftActor.waitForRecoveryComplete();
- ApplySnapshot applySnapshot = new ApplySnapshot(mock(Snapshot.class));
+ ApplySnapshot applySnapshot = new ApplySnapshot(
+ Snapshot.create(null, null, 0, 0, 0, 0, 0, persistenceId, null));
doReturn(true).when(mockSupport).handleSnapshotMessage(same(applySnapshot), any(ActorRef.class));
mockRaftActor.handleCommand(applySnapshot);
- CaptureSnapshotReply captureSnapshotReply = new CaptureSnapshotReply(ByteState.empty(),
- java.util.Optional.empty());
+ CaptureSnapshotReply captureSnapshotReply = new CaptureSnapshotReply(ByteState.empty(), Optional.empty());
doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshotReply), any(ActorRef.class));
mockRaftActor.handleCommand(captureSnapshotReply);
new MockRaftActorContext.MockPayload("foo-3"),
new MockRaftActorContext.MockPayload("foo-4")));
- leaderActor.getRaftActorContext().getSnapshotManager().persist(snapshotState, java.util.Optional.empty(),
+ leaderActor.getRaftActorContext().getSnapshotManager().persist(snapshotState, Optional.empty(),
Runtime.getRuntime().totalMemory());
assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
new MockRaftActorContext.MockPayload("foo-3"),
new MockRaftActorContext.MockPayload("foo-4")));
followerActor.handleCommand(new CaptureSnapshotReply(ByteState.of(snapshotBytes.toByteArray()),
- java.util.Optional.empty()));
+ Optional.empty()));
assertTrue(followerActor.getRaftActorContext().getSnapshotManager().isCapturing());
// The commit is needed to complete the snapshot creation process
new MockRaftActorContext.MockPayload("foo-3"),
new MockRaftActorContext.MockPayload("foo-4")));
leaderActor.handleCommand(new CaptureSnapshotReply(ByteState.of(snapshotBytes.toByteArray()),
- java.util.Optional.empty()));
+ Optional.empty()));
assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
assertEquals("Real snapshot didn't clear the log till replicatedToAllIndex", 0,
// Now send a CaptureSnapshotReply
mockActorRef.tell(new CaptureSnapshotReply(ByteState.of(fromObject("foo").toByteArray()),
- java.util.Optional.empty()), mockActorRef);
+ Optional.empty()), mockActorRef);
// Trimming log in this scenario is a no-op
assertEquals(-1, leaderActor.getReplicatedLog().getSnapshotIndex());
// Now send a CaptureSnapshotReply
mockActorRef.tell(new CaptureSnapshotReply(ByteState.of(fromObject("foo").toByteArray()),
- java.util.Optional.empty()), mockActorRef);
+ Optional.empty()), mockActorRef);
// Trimming log in this scenario is a no-op
assertEquals(3, leaderActor.getReplicatedLog().getSnapshotIndex());
ArgumentCaptor<ActorRef> replyActor = ArgumentCaptor.forClass(ActorRef.class);
verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(replyActor.capture(),
- eq(java.util.Optional.empty()));
+ eq(Optional.empty()));
byte[] stateSnapshot = new byte[]{1,2,3};
- replyActor.getValue().tell(new CaptureSnapshotReply(ByteState.of(stateSnapshot), java.util.Optional.empty()),
+ replyActor.getValue().tell(new CaptureSnapshotReply(ByteState.of(stateSnapshot), Optional.empty()),
ActorRef.noSender());
GetSnapshotReply reply = kit.expectMsgClass(GetSnapshotReply.class);
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();
+ }
}
Code Review / controller.git / blobdiff