import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.NonPersistentDataProvider;
-import org.opendaylight.controller.cluster.datastore.DataPersistenceProviderMonitor;
import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
import org.opendaylight.controller.cluster.notifications.RoleChanged;
-import org.opendaylight.controller.cluster.raft.RaftActor.DeleteEntries;
import org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
+import org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries;
import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
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.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
-import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
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.FiniteDuration;
public class RaftActorTest extends AbstractActorTest {
+ static final Logger TEST_LOG = LoggerFactory.getLogger(RaftActorTest.class);
+
private TestActorFactory factory;
@Before
@Test
public void testRaftActorRecoveryWithPersistenceEnabled() throws Exception {
+ TEST_LOG.info("testRaftActorRecoveryWithPersistenceEnabled starting");
+
new JavaTestKit(getSystem()) {{
String persistenceId = factory.generateActorId("follower-");
// log entry.
config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
+ ImmutableMap<String, String> peerAddresses = ImmutableMap.<String, String>builder().put("member1", "address").build();
ActorRef followerActor = factory.createActor(MockRaftActor.props(persistenceId,
- ImmutableMap.<String, String>builder().put("member1", "address").build(),
- Optional.<ConfigParams>of(config)), persistenceId);
+ peerAddresses, Optional.<ConfigParams>of(config)), persistenceId);
watch(followerActor);
//reinstate the actor
TestActorRef<MockRaftActor> ref = factory.createTestActor(
- MockRaftActor.props(persistenceId, Collections.<String, String>emptyMap(),
- Optional.<ConfigParams>of(config)));
+ MockRaftActor.props(persistenceId, peerAddresses, Optional.<ConfigParams>of(config)));
MockRaftActor mockRaftActor = ref.underlyingActor();
assertEquals("getRaftState", RaftState.Follower, mockRaftActor.getRaftState());
}};
+
+ TEST_LOG.info("testRaftActorRecoveryWithPersistenceEnabled ending");
}
@Test
DeleteEntries deleteEntries = new DeleteEntries(1);
mockRaftActor.handleRecover(deleteEntries);
+ org.opendaylight.controller.cluster.raft.RaftActor.DeleteEntries deprecatedDeleteEntries =
+ new org.opendaylight.controller.cluster.raft.RaftActor.DeleteEntries(1);
+ mockRaftActor.handleRecover(deprecatedDeleteEntries);
+
UpdateElectionTerm updateElectionTerm = new UpdateElectionTerm(5, "member2");
mockRaftActor.handleRecover(updateElectionTerm);
verify(mockSupport).handleRecoveryMessage(same(applyJournalEntries));
verify(mockSupport).handleRecoveryMessage(same(applyLogEntries));
verify(mockSupport).handleRecoveryMessage(same(deleteEntries));
+ verify(mockSupport).handleRecoveryMessage(same(deprecatedDeleteEntries));
verify(mockSupport).handleRecoveryMessage(same(updateElectionTerm));
}
doReturn(true).when(mockSupport).handleSnapshotMessage(same(applySnapshot));
mockRaftActor.handleCommand(applySnapshot);
- CaptureSnapshot captureSnapshot = new CaptureSnapshot(1, 1, 1, 1, 0, 1);
+ CaptureSnapshot captureSnapshot = new CaptureSnapshot(1, 1, 1, 1, 0, 1, null);
doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshot));
mockRaftActor.handleCommand(captureSnapshot);
verify(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT));
}
- @Test
- public void testUpdatingElectionTermCallsDataPersistence() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- String persistenceId = factory.generateActorId("leader-");
-
- DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
-
- config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
-
- CountDownLatch persistLatch = new CountDownLatch(1);
- DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
- dataPersistenceProviderMonitor.setPersistLatch(persistLatch);
-
- TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
- Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
-
- MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
-
- mockRaftActor.waitForInitializeBehaviorComplete();
-
- mockRaftActor.getRaftActorContext().getTermInformation().updateAndPersist(10, "foobar");
-
- assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS));
- }
- };
- }
-
- @Test
- public void testAddingReplicatedLogEntryCallsDataPersistence() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- String persistenceId = factory.generateActorId("leader-");
-
- DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
-
- config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
-
- DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
-
- TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
- Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
-
- MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
-
- mockRaftActor.waitForInitializeBehaviorComplete();
-
- MockRaftActorContext.MockReplicatedLogEntry logEntry = new MockRaftActorContext.MockReplicatedLogEntry(10, 10, mock(Payload.class));
-
- mockRaftActor.getRaftActorContext().getReplicatedLog().appendAndPersist(logEntry);
-
- verify(dataPersistenceProvider).persist(eq(logEntry), any(Procedure.class));
- }
- };
- }
-
- @Test
- public void testRemovingReplicatedLogEntryCallsDataPersistence() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- String persistenceId = factory.generateActorId("leader-");
-
- DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
-
- config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
-
- DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
-
- TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
- Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
-
- MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
-
- mockRaftActor.waitForInitializeBehaviorComplete();
-
- mockRaftActor.waitUntilLeader();
-
- mockRaftActor.getReplicatedLog().appendAndPersist(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class)));
-
- mockRaftActor.getRaftActorContext().getReplicatedLog().removeFromAndPersist(0);
-
- verify(dataPersistenceProvider, times(3)).persist(anyObject(), any(Procedure.class));
- }
- };
- }
-
@Test
public void testApplyJournalEntriesCallsDataPersistence() throws Exception {
new JavaTestKit(getSystem()) {
}};
}
- private ByteString fromObject(Object snapshot) throws Exception {
+ public static ByteString fromObject(Object snapshot) throws Exception {
ByteArrayOutputStream b = null;
ObjectOutputStream o = null;
try {