BUG-5280: centralize ShardSnapshot operations

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / ShardTest.java

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 2082dbcdd943473bd107d8426d7c223cd860af55..9c6eb257bfd6cc5d19229f1359e2774c2661d1eb 100644 (file)
--- 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
@@ -36,6 +36,7 @@ import com.google.common.base.Stopwatch;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.Uninterruptibles;
+import java.io.IOException;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
@@ -78,9 +79,10 @@ import org.opendaylight.controller.cluster.datastore.modification.DeleteModifica
 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
+import org.opendaylight.controller.cluster.datastore.persisted.PreBoronShardDataTreeSnapshot;
+import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshot;
 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
 import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
-import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
 import org.opendaylight.controller.cluster.raft.RaftActorContext;
@@ -104,7 +106,6 @@ import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelpe
 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
-import org.opendaylight.yangtools.util.StringIdentifier;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
@@ -130,7 +131,7 @@ public class ShardTest extends AbstractShardTest {
     public void testRegisterChangeListener() throws Exception {
         new ShardTestKit(getSystem()) {{
             final TestActorRef<Shard> shard = actorFactory.createTestActor(
-                    newShardProps(),  "testRegisterChangeListener");
+                    newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testRegisterChangeListener");
 
             waitUntilLeader(shard);
 
@@ -256,7 +257,7 @@ public class ShardTest extends AbstractShardTest {
     public void testRegisterDataTreeChangeListener() throws Exception {
         new ShardTestKit(getSystem()) {{
             final TestActorRef<Shard> shard = actorFactory.createTestActor(
-                    newShardProps(), "testRegisterDataTreeChangeListener");
+                    newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testRegisterDataTreeChangeListener");
 
             waitUntilLeader(shard);
 
@@ -426,7 +427,7 @@ public class ShardTest extends AbstractShardTest {
         final YangInstanceIdentifier root = YangInstanceIdentifier.EMPTY;
         final NormalizedNode<?,?> expected = readStore(store, root);
 
-        final Snapshot snapshot = Snapshot.create(SerializationUtils.serializeNormalizedNode(expected),
+        final Snapshot snapshot = Snapshot.create(new PreBoronShardDataTreeSnapshot(expected).serialize(),
                 Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4);
 
         shard.underlyingActor().getRaftActorSnapshotCohort().applySnapshot(snapshot.getState());
@@ -438,7 +439,8 @@ public class ShardTest extends AbstractShardTest {
 
     @Test
     public void testApplyState() throws Exception {
-        final TestActorRef<Shard> shard = actorFactory.createTestActor(newShardProps(), "testApplyState");
+        final TestActorRef<Shard> shard = actorFactory.createTestActor(
+                newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testApplyState");
 
         ShardTestKit.waitUntilLeader(shard);
 
@@ -448,8 +450,9 @@ public class ShardTest extends AbstractShardTest {
         writeMod.write(TestModel.TEST_PATH, node);
         writeMod.ready();
 
-        final ApplyState applyState = new ApplyState(null, new StringIdentifier("test"),
-            new ReplicatedLogImplEntry(1, 2, payloadForModification(source, writeMod)));
+        final TransactionIdentifier tx = nextTransactionId();
+        final ApplyState applyState = new ApplyState(null, tx,
+            new ReplicatedLogImplEntry(1, 2, payloadForModification(source, writeMod, tx)));
 
         shard.tell(applyState, shard);
 
@@ -478,7 +481,8 @@ public class ShardTest extends AbstractShardTest {
         InMemoryJournal.addEntry(shardID.toString(), 0, DUMMY_DATA);
 
         // Set up the InMemoryJournal.
-        InMemoryJournal.addEntry(shardID.toString(), 1, new ReplicatedLogImplEntry(0, 1, payloadForModification(source, writeMod)));
+        InMemoryJournal.addEntry(shardID.toString(), 1, new ReplicatedLogImplEntry(0, 1,
+            payloadForModification(source, writeMod, nextTransactionId())));
 
         final int nListEntries = 16;
         final Set<Integer> listEntryKeys = new HashSet<>();
@@ -493,8 +497,9 @@ public class ShardTest extends AbstractShardTest {
             final DataTreeModification mod = source.takeSnapshot().newModification();
             mod.merge(path, ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
             mod.ready();
+
             InMemoryJournal.addEntry(shardID.toString(), i+1, new ReplicatedLogImplEntry(i, 1,
-                payloadForModification(source, mod)));
+                payloadForModification(source, mod, nextTransactionId())));
         }
 
         InMemoryJournal.addEntry(shardID.toString(), nListEntries + 2,
@@ -919,7 +924,8 @@ public class ShardTest extends AbstractShardTest {
             };
 
             final TestActorRef<Shard> shard = actorFactory.createTestActor(
-                    Props.create(new DelegatingShardCreator(creator)), "testOnBatchedModificationsWhenNotLeader");
+                    Props.create(new DelegatingShardCreator(creator)).
+                        withDispatcher(Dispatchers.DefaultDispatcherId()), "testOnBatchedModificationsWhenNotLeader");
 
             waitUntilLeader(shard);
 
@@ -2055,7 +2061,6 @@ public class ShardTest extends AbstractShardTest {
         testCreateSnapshot(false, "testCreateSnapshotWithNonPersistentData");
     }
 
-    @SuppressWarnings("serial")
     private void testCreateSnapshot(final boolean persistent, final String shardActorName) throws Exception{
 
         final AtomicReference<CountDownLatch> latch = new AtomicReference<>(new CountDownLatch(1));
@@ -2102,7 +2107,8 @@ public class ShardTest extends AbstractShardTest {
             final Creator<Shard> creator = () -> new TestShard(newShardBuilder());
 
             final TestActorRef<Shard> shard = actorFactory.createTestActor(
-                    Props.create(new DelegatingShardCreator(creator)), shardActorName);
+                    Props.create(new DelegatingShardCreator(creator)).
+                        withDispatcher(Dispatchers.DefaultDispatcherId()), shardActorName);
 
             waitUntilLeader(shard);
             writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
@@ -2118,7 +2124,7 @@ public class ShardTest extends AbstractShardTest {
             awaitAndValidateSnapshot(expectedRoot);
         }
 
-        private void awaitAndValidateSnapshot(NormalizedNode<?,?> expectedRoot) throws InterruptedException {
+        private void awaitAndValidateSnapshot(NormalizedNode<?,?> expectedRoot) throws InterruptedException, IOException {
             assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
 
             assertTrue("Invalid saved snapshot " + savedSnapshot.get(),
@@ -2130,11 +2136,9 @@ public class ShardTest extends AbstractShardTest {
             savedSnapshot.set(null);
         }
 
-        private void verifySnapshot(final Snapshot snapshot, final NormalizedNode<?,?> expectedRoot) {
-
-            final NormalizedNode<?, ?> actual = SerializationUtils.deserializeNormalizedNode(snapshot.getState());
+        private void verifySnapshot(final Snapshot snapshot, final NormalizedNode<?,?> expectedRoot) throws IOException {
+            final NormalizedNode<?, ?> actual = ShardDataTreeSnapshot.deserialize(snapshot.getState()).getRootNode().get();
             assertEquals("Root node", expectedRoot, actual);
-
         }};
     }