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=998c198756d191df038de6f6cc75a8091fd0d1f1;hb=34bc6ec632529a0dfe419aa7404bb42a456fbc96;hp=9b099c2abac8223529b750c6ea906925105ec487;hpb=4019c9fd2ad99628dd790acc1ad4c104f48b6428;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 9b099c2aba..998c198756 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 @@ -9,12 +9,21 @@ import akka.japi.Creator; import akka.testkit.JavaTestKit; import akka.testkit.TestActorRef; import com.google.protobuf.ByteString; +import org.junit.After; import org.junit.Test; +import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries; import org.opendaylight.controller.cluster.raft.client.messages.FindLeader; import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply; +import org.opendaylight.controller.cluster.raft.utils.MockAkkaJournal; import org.opendaylight.controller.cluster.raft.utils.MockSnapshotStore; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; @@ -25,13 +34,21 @@ import static junit.framework.TestCase.assertEquals; public class RaftActorTest extends AbstractActorTest { + @After + public void tearDown() { + MockAkkaJournal.clearJournal(); + MockSnapshotStore.setMockSnapshot(null); + } + public static class MockRaftActor extends RaftActor { - boolean applySnapshotCalled = false; + private boolean applySnapshotCalled = false; + private List state; public MockRaftActor(String id, Map peerAddresses) { super(id, peerAddresses); + state = new ArrayList<>(); } public RaftActorContext getRaftActorContext() { @@ -42,6 +59,10 @@ public class RaftActorTest extends AbstractActorTest { return applySnapshotCalled; } + public List getState() { + return state; + } + public static Props props(final String id, final Map peerAddresses){ return Props.create(new Creator(){ @@ -51,9 +72,8 @@ public class RaftActorTest extends AbstractActorTest { }); } - @Override protected void applyState(ActorRef clientActor, - String identifier, - Object data) { + @Override protected void applyState(ActorRef clientActor, String identifier, Object data) { + state.add(data); } @Override protected void createSnapshot() { @@ -61,7 +81,17 @@ public class RaftActorTest extends AbstractActorTest { } @Override protected void applySnapshot(ByteString snapshot) { - applySnapshotCalled = true; + applySnapshotCalled = true; + try { + Object data = toObject(snapshot); + if (data instanceof List) { + state.addAll((List) data); + } + } catch (ClassNotFoundException e) { + e.printStackTrace(); + } catch (IOException e) { + e.printStackTrace(); + } } @Override protected void onStateChanged() { @@ -71,6 +101,26 @@ public class RaftActorTest extends AbstractActorTest { return this.getId(); } + private Object toObject(ByteString bs) throws ClassNotFoundException, IOException { + Object obj = null; + ByteArrayInputStream bis = null; + ObjectInputStream ois = null; + try { + bis = new ByteArrayInputStream(bs.toByteArray()); + ois = new ObjectInputStream(bis); + obj = ois.readObject(); + } finally { + if (bis != null) { + bis.close(); + } + if (ois != null) { + ois.close(); + } + } + return obj; + } + + } @@ -151,7 +201,7 @@ public class RaftActorTest extends AbstractActorTest { } @Test - public void testActorRecovery() { + public void testRaftActorRecovery() { new JavaTestKit(getSystem()) {{ new Within(duration("1 seconds")) { protected void run() { @@ -161,20 +211,50 @@ public class RaftActorTest extends AbstractActorTest { ActorRef followerActor = getSystem().actorOf( MockRaftActor.props(persistenceId, Collections.EMPTY_MAP), persistenceId); + List snapshotUnappliedEntries = new ArrayList<>(); + ReplicatedLogEntry entry1 = new MockRaftActorContext.MockReplicatedLogEntry(1, 4, new MockRaftActorContext.MockPayload("E")); + snapshotUnappliedEntries.add(entry1); + + int lastAppliedDuringSnapshotCapture = 3; + int lastIndexDuringSnapshotCapture = 4; + ByteString snapshotBytes = null; + try { + // 4 messages as part of snapshot, which are applied to state + snapshotBytes = fromObject(Arrays.asList(new MockRaftActorContext.MockPayload("A"), + new MockRaftActorContext.MockPayload("B"), + new MockRaftActorContext.MockPayload("C"), + new MockRaftActorContext.MockPayload("D"))); + } catch (Exception e) { + e.printStackTrace(); + } + Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(), + snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1 , + lastAppliedDuringSnapshotCapture, 1); + MockSnapshotStore.setMockSnapshot(snapshot); + MockSnapshotStore.setPersistenceId(persistenceId); + + // add more entries after snapshot is taken List entries = new ArrayList<>(); - ReplicatedLogEntry entry1 = new MockRaftActorContext.MockReplicatedLogEntry(1, 4, new MockRaftActorContext.MockPayload("E")); ReplicatedLogEntry entry2 = new MockRaftActorContext.MockReplicatedLogEntry(1, 5, new MockRaftActorContext.MockPayload("F")); - entries.add(entry1); + ReplicatedLogEntry entry3 = new MockRaftActorContext.MockReplicatedLogEntry(1, 6, new MockRaftActorContext.MockPayload("G")); + ReplicatedLogEntry entry4 = new MockRaftActorContext.MockReplicatedLogEntry(1, 7, new MockRaftActorContext.MockPayload("H")); entries.add(entry2); + entries.add(entry3); + entries.add(entry4); - int lastApplied = 3; - int lastIndex = 5; - Snapshot snapshot = Snapshot.create("A B C D".getBytes(), entries, lastIndex, 1 , lastApplied, 1); - MockSnapshotStore.setMockSnapshot(snapshot); - MockSnapshotStore.setPersistenceId(persistenceId); + int lastAppliedToState = 5; + int lastIndex = 7; + + MockAkkaJournal.addToJournal(5, entry2); + // 2 entries are applied to state besides the 4 entries in snapshot + MockAkkaJournal.addToJournal(6, new ApplyLogEntries(lastAppliedToState)); + MockAkkaJournal.addToJournal(7, entry3); + MockAkkaJournal.addToJournal(8, entry4); + // kill the actor followerActor.tell(PoisonPill.getInstance(), null); + try { // give some time for actor to die Thread.sleep(200); @@ -182,24 +262,47 @@ public class RaftActorTest extends AbstractActorTest { e.printStackTrace(); } - TestActorRef ref = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, Collections.EMPTY_MAP)); + //reinstate the actor + TestActorRef ref = TestActorRef.create(getSystem(), + MockRaftActor.props(persistenceId, Collections.EMPTY_MAP)); + try { //give some time for snapshot offer to get called. Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); } + RaftActorContext context = ref.underlyingActor().getRaftActorContext(); - assertEquals(entries.size(), context.getReplicatedLog().size()); - assertEquals(lastApplied, context.getLastApplied()); - assertEquals(lastApplied, context.getCommitIndex()); + assertEquals(snapshotUnappliedEntries.size() + entries.size(), context.getReplicatedLog().size()); + assertEquals(lastIndex, context.getReplicatedLog().lastIndex()); + assertEquals(lastAppliedToState, context.getLastApplied()); + assertEquals(lastAppliedToState, context.getCommitIndex()); assertTrue(ref.underlyingActor().isApplySnapshotCalled()); + assertEquals(6, ref.underlyingActor().getState().size()); } - }; }}; } - + private ByteString fromObject(Object snapshot) throws Exception { + ByteArrayOutputStream b = null; + ObjectOutputStream o = null; + try { + b = new ByteArrayOutputStream(); + o = new ObjectOutputStream(b); + o.writeObject(snapshot); + byte[] snapshotBytes = b.toByteArray(); + return ByteString.copyFrom(snapshotBytes); + } finally { + if (o != null) { + o.flush(); + o.close(); + } + if (b != null) { + b.close(); + } + } + } }