X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardTest.java;h=14fc3a12bd9b97bf891586eb3902c502fa18e142;hb=3f2221486de63178fbfbb43508ce9466c0b23b73;hp=03a18ea6c38c39600f2c4bd3a49b848ace89a09d;hpb=a1878a52f1d6df37acae733430911fbce5be1d50;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java index 03a18ea6c3..14fc3a12bd 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java @@ -1,5 +1,16 @@ package org.opendaylight.controller.cluster.datastore; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.opendaylight.controller.cluster.datastore.DataStoreVersions.CURRENT_VERSION; import akka.actor.ActorRef; import akka.actor.PoisonPill; import akka.actor.Props; @@ -16,12 +27,23 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.Uninterruptibles; +import java.io.IOException; +import java.util.Collections; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.junit.After; import org.junit.Before; import org.junit.Test; import org.mockito.InOrder; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier; import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; @@ -38,6 +60,7 @@ import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeList import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; import org.opendaylight.controller.cluster.datastore.modification.MergeModification; import org.opendaylight.controller.cluster.datastore.modification.Modification; +import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload; import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification; import org.opendaylight.controller.cluster.datastore.modification.WriteModification; import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec; @@ -54,8 +77,8 @@ import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot; import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; import org.opendaylight.controller.cluster.raft.client.messages.FindLeader; import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply; +import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload; -import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; import org.opendaylight.controller.md.cluster.datastore.model.TestModel; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker; @@ -79,25 +102,6 @@ import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.Await; import scala.concurrent.Future; import scala.concurrent.duration.FiniteDuration; -import java.io.IOException; -import java.util.Collections; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.inOrder; public class ShardTest extends AbstractActorTest { @@ -108,14 +112,13 @@ public class ShardTest extends AbstractActorTest { private final ShardIdentifier shardID = ShardIdentifier.builder().memberName("member-1") .shardName("inventory").type("config" + NEXT_SHARD_NUM.getAndIncrement()).build(); - private DatastoreContext dataStoreContext = DatastoreContext.newBuilder(). + private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder(). shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000). - shardHeartbeatIntervalInMillis(100).build(); + shardHeartbeatIntervalInMillis(100); @Before public void setUp() { - System.setProperty("shard.persistent", "false"); - + Builder newBuilder = DatastoreContext.newBuilder(); InMemorySnapshotStore.clear(); InMemoryJournal.clear(); } @@ -126,9 +129,13 @@ public class ShardTest extends AbstractActorTest { InMemoryJournal.clear(); } + private DatastoreContext newDatastoreContext() { + return dataStoreContextBuilder.build(); + } + private Props newShardProps() { return Shard.props(shardID, Collections.emptyMap(), - dataStoreContext, SCHEMA_CONTEXT); + newDatastoreContext(), SCHEMA_CONTEXT); } @Test @@ -185,9 +192,9 @@ public class ShardTest extends AbstractActorTest { @Override public Shard create() throws Exception { return new Shard(shardID, Collections.emptyMap(), - dataStoreContext, SCHEMA_CONTEXT) { + newDatastoreContext(), SCHEMA_CONTEXT) { @Override - public void onReceiveCommand(final Object message) { + public void onReceiveCommand(final Object message) throws Exception { if(message instanceof ElectionTimeout && firstElectionTimeout) { // Got the first ElectionTimeout. We don't forward it to the // base Shard yet until we've sent the RegisterChangeListener @@ -305,14 +312,15 @@ public class ShardTest extends AbstractActorTest { }}; } + @SuppressWarnings("serial") @Test - public void testPeerAddressResolved(){ + public void testPeerAddressResolved() throws Exception { new ShardTestKit(getSystem()) {{ final CountDownLatch recoveryComplete = new CountDownLatch(1); class TestShard extends Shard { TestShard() { super(shardID, Collections.singletonMap(shardID, null), - dataStoreContext, SCHEMA_CONTEXT); + newDatastoreContext(), SCHEMA_CONTEXT); } Map getPeerAddresses() { @@ -352,7 +360,7 @@ public class ShardTest extends AbstractActorTest { } @Test - public void testApplySnapshot() throws ExecutionException, InterruptedException { + public void testApplySnapshot() throws Exception { TestActorRef shard = TestActorRef.create(getSystem(), newShardProps(), "testApplySnapshot"); @@ -386,11 +394,26 @@ public class ShardTest extends AbstractActorTest { NormalizedNode node = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - MutableCompositeModification compMod = new MutableCompositeModification(); - compMod.addModification(new WriteModification(TestModel.TEST_PATH, node, SCHEMA_CONTEXT)); - Payload payload = new CompositeModificationPayload(compMod.toSerializable()); - ApplyState applyState = new ApplyState(null, "test", - new ReplicatedLogImplEntry(1, 2, payload)); + ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2, + newModificationPayload(new WriteModification(TestModel.TEST_PATH, node)))); + + shard.underlyingActor().onReceiveCommand(applyState); + + NormalizedNode actual = readStore(shard, TestModel.TEST_PATH); + assertEquals("Applied state", node, actual); + + shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); + } + + @Test + public void testApplyStateLegacy() throws Exception { + + TestActorRef shard = TestActorRef.create(getSystem(), newShardProps(), "testApplyStateLegacy"); + + NormalizedNode node = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2, + newLegacyByteStringPayload(new WriteModification(TestModel.TEST_PATH, node)))); shard.underlyingActor().onReceiveCommand(applyState); @@ -426,22 +449,45 @@ public class ShardTest extends AbstractActorTest { // Set up the InMemoryJournal. - InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, newPayload( + InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, newLegacyPayload( new WriteModification(TestModel.OUTER_LIST_PATH, - ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), - SCHEMA_CONTEXT)))); + ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build())))); - int nListEntries = 11; + int nListEntries = 16; Set listEntryKeys = new HashSet<>(); - for(int i = 1; i <= nListEntries; i++) { + int i = 1; + + // Add some of the legacy CompositeModificationPayload + for(; i <= 2; i++) { listEntryKeys.add(Integer.valueOf(i)); YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build(); Modification mod = new MergeModification(path, - ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i), - SCHEMA_CONTEXT); + ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i)); InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1, - newPayload(mod))); + newLegacyPayload(mod))); + } + + // Add some of the legacy CompositeModificationByteStringPayload + for(; i <= 5; i++) { + listEntryKeys.add(Integer.valueOf(i)); + YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) + .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build(); + Modification mod = new MergeModification(path, + ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i)); + InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1, + newLegacyByteStringPayload(mod))); + } + + // Add some of the ModificationPayload + for(; i <= nListEntries; i++) { + listEntryKeys.add(Integer.valueOf(i)); + YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) + .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build(); + Modification mod = new MergeModification(path, + ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i)); + InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1, + newModificationPayload(mod))); } InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1, @@ -455,7 +501,7 @@ public class ShardTest extends AbstractActorTest { @Override public Shard create() throws Exception { return new Shard(shardID, Collections.emptyMap(), - dataStoreContext, SCHEMA_CONTEXT) { + newDatastoreContext(), SCHEMA_CONTEXT) { @Override protected void onRecoveryComplete() { try { @@ -506,7 +552,7 @@ public class ShardTest extends AbstractActorTest { shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); } - private CompositeModificationPayload newPayload(Modification... mods) { + private CompositeModificationPayload newLegacyPayload(final Modification... mods) { MutableCompositeModification compMod = new MutableCompositeModification(); for(Modification mod: mods) { compMod.addModification(mod); @@ -515,15 +561,33 @@ public class ShardTest extends AbstractActorTest { return new CompositeModificationPayload(compMod.toSerializable()); } - private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(String cohortName, - InMemoryDOMDataStore dataStore, YangInstanceIdentifier path, NormalizedNode data, - MutableCompositeModification modification) { + private CompositeModificationByteStringPayload newLegacyByteStringPayload(final Modification... mods) { + MutableCompositeModification compMod = new MutableCompositeModification(); + for(Modification mod: mods) { + compMod.addModification(mod); + } + + return new CompositeModificationByteStringPayload(compMod.toSerializable()); + } + + private ModificationPayload newModificationPayload(final Modification... mods) throws IOException { + MutableCompositeModification compMod = new MutableCompositeModification(); + for(Modification mod: mods) { + compMod.addModification(mod); + } + + return new ModificationPayload(compMod); + } + + private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName, + final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode data, + final MutableCompositeModification modification) { return setupMockWriteTransaction(cohortName, dataStore, path, data, modification, null); } - private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(String cohortName, - InMemoryDOMDataStore dataStore, YangInstanceIdentifier path, NormalizedNode data, - MutableCompositeModification modification, + private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName, + final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode data, + final MutableCompositeModification modification, final Function> preCommit) { DOMStoreWriteTransaction tx = dataStore.newWriteOnlyTransaction(); @@ -533,14 +597,14 @@ public class ShardTest extends AbstractActorTest { doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) { + public ListenableFuture answer(final InvocationOnMock invocation) { return realCohort.canCommit(); } }).when(cohort).canCommit(); doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) throws Throwable { + public ListenableFuture answer(final InvocationOnMock invocation) throws Throwable { if(preCommit != null) { return preCommit.apply(realCohort); } else { @@ -551,19 +615,19 @@ public class ShardTest extends AbstractActorTest { doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) throws Throwable { + public ListenableFuture answer(final InvocationOnMock invocation) throws Throwable { return realCohort.commit(); } }).when(cohort).commit(); doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) throws Throwable { + public ListenableFuture answer(final InvocationOnMock invocation) throws Throwable { return realCohort.abort(); } }).when(cohort).abort(); - modification.addModification(new WriteModification(path, data, SCHEMA_CONTEXT)); + modification.addModification(new WriteModification(path, data)); return cohort; } @@ -571,7 +635,6 @@ public class ShardTest extends AbstractActorTest { @SuppressWarnings({ "unchecked" }) @Test public void testConcurrentThreePhaseCommits() throws Throwable { - System.setProperty("shard.persistent", "true"); new ShardTestKit(getSystem()) {{ final TestActorRef shard = TestActorRef.create(getSystem(), newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), @@ -610,7 +673,8 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction message for the first Tx that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID1, cohort1, modification1, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, + cohort1, modification1, true), getRef()); ReadyTransactionReply readyReply = ReadyTransactionReply.fromSerializable( expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS)); assertEquals("Cohort path", shard.path().toString(), readyReply.getCohortPath()); @@ -624,10 +688,12 @@ public class ShardTest extends AbstractActorTest { // Send the ForwardedReadyTransaction for the next 2 Tx's. - shard.tell(new ForwardedReadyTransaction(transactionID2, cohort2, modification2, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, + cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID3, cohort3, modification3, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION, + cohort3, modification3, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); // Send the CanCommitTransaction message for the next 2 Tx's. These should get queued and @@ -653,12 +719,12 @@ public class ShardTest extends AbstractActorTest { class OnFutureComplete extends OnComplete { private final Class expRespType; - OnFutureComplete(Class expRespType) { + OnFutureComplete(final Class expRespType) { this.expRespType = expRespType; } @Override - public void onComplete(Throwable error, Object resp) { + public void onComplete(final Throwable error, final Object resp) { if(error != null) { caughtEx.set(new AssertionError(getClass().getSimpleName() + " failure", error)); } else { @@ -671,7 +737,7 @@ public class ShardTest extends AbstractActorTest { } } - void onSuccess(Object resp) throws Exception { + void onSuccess(final Object resp) throws Exception { } } @@ -681,7 +747,7 @@ public class ShardTest extends AbstractActorTest { } @Override - public void onComplete(Throwable error, Object resp) { + public void onComplete(final Throwable error, final Object resp) { super.onComplete(error, resp); commitLatch.countDown(); } @@ -690,13 +756,13 @@ public class ShardTest extends AbstractActorTest { class OnCanCommitFutureComplete extends OnFutureComplete { private final String transactionID; - OnCanCommitFutureComplete(String transactionID) { + OnCanCommitFutureComplete(final String transactionID) { super(CanCommitTransactionReply.SERIALIZABLE_CLASS); this.transactionID = transactionID; } @Override - void onSuccess(Object resp) throws Exception { + void onSuccess(final Object resp) throws Exception { CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(resp); assertEquals("Can commit", true, canCommitReply.getCanCommit()); @@ -747,15 +813,72 @@ public class ShardTest extends AbstractActorTest { assertTrue("Missing leaf " + TestModel.ID_QNAME.getLocalName(), idLeaf.isPresent()); assertEquals(TestModel.ID_QNAME.getLocalName() + " value", 1, idLeaf.get().getValue()); - for(int i = 0; i < 20 * 5; i++) { - long lastLogIndex = shard.underlyingActor().getShardMBean().getLastLogIndex(); - if(lastLogIndex == 2) { - break; - } - Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + verifyLastLogIndex(shard, 2); + + shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); + }}; + } + + private void verifyLastLogIndex(TestActorRef shard, long expectedValue) { + for(int i = 0; i < 20 * 5; i++) { + long lastLogIndex = shard.underlyingActor().getShardMBean().getLastLogIndex(); + if(lastLogIndex == expectedValue) { + break; } + Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + } + + assertEquals("Last log index", expectedValue, shard.underlyingActor().getShardMBean().getLastLogIndex()); + } + + @Test + public void testCommitWithPersistenceDisabled() throws Throwable { + dataStoreContextBuilder.persistent(false); + new ShardTestKit(getSystem()) {{ + final TestActorRef shard = TestActorRef.create(getSystem(), + newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), + "testCommitPhaseFailure"); + + waitUntilLeader(shard); + + InMemoryDOMDataStore dataStore = shard.underlyingActor().getDataStore(); + + // Setup a simulated transactions with a mock cohort. + + String transactionID = "tx"; + MutableCompositeModification modification = new MutableCompositeModification(); + NormalizedNode containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + DOMStoreThreePhaseCommitCohort cohort = setupMockWriteTransaction("cohort", dataStore, + TestModel.TEST_PATH, containerNode, modification); + + FiniteDuration duration = duration("5 seconds"); + + // Simulate the ForwardedReadyTransaction messages that would be sent + // by the ShardTransaction. + + shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, + cohort, modification, true), getRef()); + expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); + + // Send the CanCommitTransaction message. + + shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef()); + CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable( + expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS)); + assertEquals("Can commit", true, canCommitReply.getCanCommit()); + + // Send the CanCommitTransaction message. + + shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef()); + expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS); + + InOrder inOrder = inOrder(cohort); + inOrder.verify(cohort).canCommit(); + inOrder.verify(cohort).preCommit(); + inOrder.verify(cohort).commit(); - assertEquals("Last log index", 2, shard.underlyingActor().getShardMBean().getLastLogIndex()); + NormalizedNode actualNode = readStore(shard, TestModel.TEST_PATH); + assertEquals(TestModel.TEST_QNAME.getLocalName(), containerNode, actualNode); shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); }}; @@ -791,10 +914,12 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID1, cohort1, modification1, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, + cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, cohort2, modification2, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, + cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); // Send the CanCommitTransaction message for the first Tx. @@ -821,7 +946,7 @@ public class ShardTest extends AbstractActorTest { final CountDownLatch latch = new CountDownLatch(1); canCommitFuture.onComplete(new OnComplete() { @Override - public void onComplete(Throwable t, Object resp) { + public void onComplete(final Throwable t, final Object resp) { latch.countDown(); } }, getSystem().dispatcher()); @@ -858,7 +983,8 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID, cohort, modification, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, + cohort, modification, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); // Send the CanCommitTransaction message. @@ -901,7 +1027,8 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID, cohort, modification, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, + cohort, modification, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); // Send the CanCommitTransaction message. @@ -915,7 +1042,6 @@ public class ShardTest extends AbstractActorTest { @Test public void testAbortBeforeFinishCommit() throws Throwable { - System.setProperty("shard.persistent", "true"); new ShardTestKit(getSystem()) {{ final TestActorRef shard = TestActorRef.create(getSystem(), newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), @@ -924,26 +1050,24 @@ public class ShardTest extends AbstractActorTest { waitUntilLeader(shard); final FiniteDuration duration = duration("5 seconds"); - final Timeout timeout = new Timeout(duration); - InMemoryDOMDataStore dataStore = shard.underlyingActor().getDataStore(); final String transactionID = "tx1"; - final CountDownLatch abortComplete = new CountDownLatch(1); Function> preCommit = new Function>() { @Override public ListenableFuture apply(final DOMStoreThreePhaseCommitCohort cohort) { ListenableFuture preCommitFuture = cohort.preCommit(); - Future abortFuture = Patterns.ask(shard, - new AbortTransaction(transactionID).toSerializable(), timeout); - abortFuture.onComplete(new OnComplete() { - @Override - public void onComplete(Throwable e, Object resp) { - abortComplete.countDown(); - } - }, getSystem().dispatcher()); + // Simulate an AbortTransaction message occurring during replication, after + // persisting and before finishing the commit to the in-memory store. + // We have no followers so due to optimizations in the RaftActor, it does not + // attempt replication and thus we can't send an AbortTransaction message b/c + // it would be processed too late after CommitTransaction completes. So we'll + // simulate an AbortTransaction message occurring during replication by calling + // the shard directly. + // + shard.underlyingActor().doAbortTransaction(transactionID, null); return preCommitFuture; } @@ -954,7 +1078,8 @@ public class ShardTest extends AbstractActorTest { TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification, preCommit); - shard.tell(new ForwardedReadyTransaction(transactionID, cohort, modification, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, + cohort, modification, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef()); @@ -962,14 +1087,14 @@ public class ShardTest extends AbstractActorTest { expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS)); assertEquals("Can commit", true, canCommitReply.getCanCommit()); - Future commitFuture = Patterns.ask(shard, - new CommitTransaction(transactionID).toSerializable(), timeout); - - assertEquals("Abort complete", true, abortComplete.await(5, TimeUnit.SECONDS)); - - Await.result(commitFuture, duration); + shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef()); + expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS); NormalizedNode node = readStore(shard, TestModel.TEST_PATH); + + // Since we're simulating an abort occurring during replication and before finish commit, + // the data should still get written to the in-memory store since we've gotten past + // canCommit and preCommit and persisted the data. assertNotNull(TestModel.TEST_QNAME.getLocalName() + " not found", node); shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); @@ -978,7 +1103,7 @@ public class ShardTest extends AbstractActorTest { @Test public void testTransactionCommitTimeout() throws Throwable { - dataStoreContext = DatastoreContext.newBuilder().shardTransactionCommitTimeoutInSeconds(1).build(); + dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1); new ShardTestKit(getSystem()) {{ final TestActorRef shard = TestActorRef.create(getSystem(), @@ -1018,10 +1143,12 @@ public class ShardTest extends AbstractActorTest { // Ready the Tx's - shard.tell(new ForwardedReadyTransaction(transactionID1, cohort1, modification1, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, + cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, cohort2, modification2, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, + cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); // canCommit 1st Tx. We don't send the commit so it should timeout. @@ -1048,7 +1175,7 @@ public class ShardTest extends AbstractActorTest { @Test public void testTransactionCommitQueueCapacityExceeded() throws Throwable { - dataStoreContext = DatastoreContext.newBuilder().shardTransactionCommitQueueCapacity(1).build(); + dataStoreContextBuilder.shardTransactionCommitQueueCapacity(1); new ShardTestKit(getSystem()) {{ final TestActorRef shard = TestActorRef.create(getSystem(), @@ -1080,13 +1207,16 @@ public class ShardTest extends AbstractActorTest { // Ready the Tx's - shard.tell(new ForwardedReadyTransaction(transactionID1, cohort1, modification1, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, + cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, cohort2, modification2, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, + cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID3, cohort3, modification3, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION, + cohort3, modification3, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); // canCommit 1st Tx. @@ -1149,10 +1279,12 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID1, cohort1, modification1, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, + cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, cohort2, modification2, true), getRef()); + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, + cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); // Send the CanCommitTransaction message for the first Tx. @@ -1176,15 +1308,7 @@ public class ShardTest extends AbstractActorTest { // Wait for the 2nd Tx to complete the canCommit phase. - final CountDownLatch latch = new CountDownLatch(1); - canCommitFuture.onComplete(new OnComplete() { - @Override - public void onComplete(Throwable t, Object resp) { - latch.countDown(); - } - }, getSystem().dispatcher()); - - assertEquals("2nd CanCommit complete", true, latch.await(5, TimeUnit.SECONDS)); + Await.ready(canCommitFuture, duration); InOrder inOrder = inOrder(cohort1, cohort2); inOrder.verify(cohort1).canCommit(); @@ -1196,16 +1320,29 @@ public class ShardTest extends AbstractActorTest { @Test public void testCreateSnapshot() throws IOException, InterruptedException { + testCreateSnapshot(true, "testCreateSnapshot"); + } + + @Test + public void testCreateSnapshotWithNonPersistentData() throws IOException, InterruptedException { + testCreateSnapshot(false, "testCreateSnapshotWithNonPersistentData"); + } + + @SuppressWarnings("serial") + public void testCreateSnapshot(final boolean persistent, final String shardActorName) throws IOException, InterruptedException { + final DatastoreContext dataStoreContext = DatastoreContext.newBuilder(). + shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(persistent).build(); + new ShardTestKit(getSystem()) {{ final AtomicReference latch = new AtomicReference<>(new CountDownLatch(1)); Creator creator = new Creator() { @Override public Shard create() throws Exception { return new Shard(shardID, Collections.emptyMap(), - dataStoreContext, SCHEMA_CONTEXT) { + newDatastoreContext(), SCHEMA_CONTEXT) { @Override - public void saveSnapshot(Object snapshot) { - super.saveSnapshot(snapshot); + protected void commitSnapshot(final long sequenceNumber) { + super.commitSnapshot(sequenceNumber); latch.get().countDown(); } }; @@ -1213,7 +1350,7 @@ public class ShardTest extends AbstractActorTest { }; TestActorRef shard = TestActorRef.create(getSystem(), - Props.create(new DelegatingShardCreator(creator)), "testCreateSnapshot"); + Props.create(new DelegatingShardCreator(creator)), shardActorName); waitUntilLeader(shard); @@ -1236,8 +1373,7 @@ public class ShardTest extends AbstractActorTest { */ @Test public void testInMemoryDataStoreRestore() throws ReadFailedException { - InMemoryDOMDataStore store = new InMemoryDOMDataStore("test", MoreExecutors.listeningDecorator( - MoreExecutors.sameThreadExecutor()), MoreExecutors.sameThreadExecutor()); + InMemoryDOMDataStore store = new InMemoryDOMDataStore("test", MoreExecutors.sameThreadExecutor()); store.onGlobalContextUpdated(SCHEMA_CONTEXT); @@ -1247,7 +1383,7 @@ public class ShardTest extends AbstractActorTest { commitTransaction(putTransaction); - NormalizedNode expected = readStore(store); + NormalizedNode expected = readStore(store); DOMStoreWriteTransaction writeTransaction = store.newWriteOnlyTransaction(); @@ -1256,13 +1392,47 @@ public class ShardTest extends AbstractActorTest { commitTransaction(writeTransaction); - NormalizedNode actual = readStore(store); + NormalizedNode actual = readStore(store); assertEquals(expected, actual); + } + + @Test + public void testRecoveryApplicable(){ + + final DatastoreContext persistentContext = DatastoreContext.newBuilder(). + shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(true).build(); + + final Props persistentProps = Shard.props(shardID, Collections.emptyMap(), + persistentContext, SCHEMA_CONTEXT); + + final DatastoreContext nonPersistentContext = DatastoreContext.newBuilder(). + shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(false).build(); + + final Props nonPersistentProps = Shard.props(shardID, Collections.emptyMap(), + nonPersistentContext, SCHEMA_CONTEXT); + + new ShardTestKit(getSystem()) {{ + TestActorRef shard1 = TestActorRef.create(getSystem(), + persistentProps, "testPersistence1"); + + assertTrue("Recovery Applicable", shard1.underlyingActor().getDataPersistenceProvider().isRecoveryApplicable()); + + shard1.tell(PoisonPill.getInstance(), ActorRef.noSender()); + + TestActorRef shard2 = TestActorRef.create(getSystem(), + nonPersistentProps, "testPersistence2"); + + assertFalse("Recovery Not Applicable", shard2.underlyingActor().getDataPersistenceProvider().isRecoveryApplicable()); + + shard2.tell(PoisonPill.getInstance(), ActorRef.noSender()); + + }}; } - private NormalizedNode readStore(InMemoryDOMDataStore store) throws ReadFailedException { + + private NormalizedNode readStore(final InMemoryDOMDataStore store) throws ReadFailedException { DOMStoreReadTransaction transaction = store.newReadOnlyTransaction(); CheckedFuture>, ReadFailedException> read = transaction.read(YangInstanceIdentifier.builder().build()); @@ -1276,7 +1446,7 @@ public class ShardTest extends AbstractActorTest { return normalizedNode; } - private void commitTransaction(DOMStoreWriteTransaction transaction) { + private void commitTransaction(final DOMStoreWriteTransaction transaction) { DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready(); ListenableFuture future = commitCohort.preCommit(); @@ -1292,13 +1462,13 @@ public class ShardTest extends AbstractActorTest { return new AsyncDataChangeListener>() { @Override public void onDataChanged( - AsyncDataChangeEvent> change) { + final AsyncDataChangeEvent> change) { } }; } - private NormalizedNode readStore(TestActorRef shard, YangInstanceIdentifier id) + static NormalizedNode readStore(final TestActorRef shard, final YangInstanceIdentifier id) throws ExecutionException, InterruptedException { DOMStoreReadTransaction transaction = shard.underlyingActor().getDataStore().newReadOnlyTransaction(); @@ -1313,7 +1483,7 @@ public class ShardTest extends AbstractActorTest { return node; } - private void writeToStore(TestActorRef shard, YangInstanceIdentifier id, NormalizedNode node) + private void writeToStore(final TestActorRef shard, final YangInstanceIdentifier id, final NormalizedNode node) throws ExecutionException, InterruptedException { DOMStoreWriteTransaction transaction = shard.underlyingActor().getDataStore().newWriteOnlyTransaction(); @@ -1324,10 +1494,11 @@ public class ShardTest extends AbstractActorTest { commitCohort.commit().get(); } + @SuppressWarnings("serial") private static final class DelegatingShardCreator implements Creator { private final Creator delegate; - DelegatingShardCreator(Creator delegate) { + DelegatingShardCreator(final Creator delegate) { this.delegate = delegate; }