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

   1 package org.opendaylight.controller.cluster.datastore;
   2
   3 import static org.junit.Assert.assertEquals;
   4 import static org.junit.Assert.assertFalse;
   5 import static org.junit.Assert.assertNotNull;
   6 import static org.junit.Assert.assertNull;
   7 import static org.junit.Assert.assertTrue;
   8 import static org.mockito.Mockito.doReturn;
   9 import static org.mockito.Mockito.inOrder;
  10 import static org.mockito.Mockito.mock;
  11 import static org.mockito.Mockito.reset;
  12 import static org.opendaylight.controller.cluster.datastore.DataStoreVersions.CURRENT_VERSION;
  13 import akka.actor.ActorRef;
  14 import akka.actor.ActorSelection;
  15 import akka.actor.PoisonPill;
  16 import akka.actor.Props;
  17 import akka.dispatch.Dispatchers;
  18 import akka.dispatch.OnComplete;
  19 import akka.japi.Creator;
  20 import akka.pattern.Patterns;
  21 import akka.persistence.SaveSnapshotSuccess;
  22 import akka.testkit.TestActorRef;
  23 import akka.util.Timeout;
  24 import com.google.common.base.Function;
  25 import com.google.common.base.Optional;
  26 import com.google.common.util.concurrent.Futures;
  27 import com.google.common.util.concurrent.ListenableFuture;
  28 import com.google.common.util.concurrent.Uninterruptibles;
  29 import java.io.IOException;
  30 import java.util.Collections;
  31 import java.util.HashSet;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.Set;
  35 import java.util.concurrent.CountDownLatch;
  36 import java.util.concurrent.TimeUnit;
  37 import java.util.concurrent.atomic.AtomicBoolean;
  38 import java.util.concurrent.atomic.AtomicReference;
  39 import org.junit.Test;
  40 import org.mockito.InOrder;
  41 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  42 import org.opendaylight.controller.cluster.DelegatingPersistentDataProvider;
  43 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  44 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
  45 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
  46 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  47 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
  48 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  49 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
  50 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  51 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  52 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  53 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
  54 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
  55 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  56 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
  57 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  58 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
  59 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
  60 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  61 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
  62 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  63 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  64 import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload;
  65 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  66 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  67 import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
  68 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
  69 import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
  70 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
  71 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
  72 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  73 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  74 import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
  75 import org.opendaylight.controller.cluster.raft.Snapshot;
  76 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
  77 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
  78 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
  79 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
  80 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
  81 import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
  82 import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
  83 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  84 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  85 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  86 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  87 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
  88 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  89 import org.opendaylight.controller.protobuff.messages.cohort3pc.ThreePhaseCommitCohortMessages;
  90 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
  91 import org.opendaylight.yangtools.yang.common.QName;
  92 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  93 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  94 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  95 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  96 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  97 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  98 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  99 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
 100 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
 101 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateTip;
 102 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
 103 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
 104 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
 105 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
 106 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
 107 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
 108 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 109 import scala.concurrent.Await;
 110 import scala.concurrent.Future;
 111 import scala.concurrent.duration.FiniteDuration;
 112
 113 public class ShardTest extends AbstractShardTest {
 114     private static final QName CARS_QNAME = QName.create("urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom:store:test:cars", "2014-03-13", "cars");
 115
 116     @Test
 117     public void testRegisterChangeListener() throws Exception {
 118         new ShardTestKit(getSystem()) {{
 119             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 120                     newShardProps(),  "testRegisterChangeListener");
 121
 122             waitUntilLeader(shard);
 123
 124             shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
 125
 126             MockDataChangeListener listener = new MockDataChangeListener(1);
 127             ActorRef dclActor = getSystem().actorOf(DataChangeListener.props(listener),
 128                     "testRegisterChangeListener-DataChangeListener");
 129
 130             shard.tell(new RegisterChangeListener(TestModel.TEST_PATH,
 131                     dclActor, AsyncDataBroker.DataChangeScope.BASE), getRef());
 132
 133             RegisterChangeListenerReply reply = expectMsgClass(duration("3 seconds"),
 134                     RegisterChangeListenerReply.class);
 135             String replyPath = reply.getListenerRegistrationPath().toString();
 136             assertTrue("Incorrect reply path: " + replyPath, replyPath.matches(
 137                     "akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
 138
 139             YangInstanceIdentifier path = TestModel.TEST_PATH;
 140             writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 141
 142             listener.waitForChangeEvents(path);
 143
 144             dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
 145             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 146         }};
 147     }
 148
 149     @SuppressWarnings("serial")
 150     @Test
 151     public void testChangeListenerNotifiedWhenNotTheLeaderOnRegistration() throws Exception {
 152         // This test tests the timing window in which a change listener is registered before the
 153         // shard becomes the leader. We verify that the listener is registered and notified of the
 154         // existing data when the shard becomes the leader.
 155         new ShardTestKit(getSystem()) {{
 156             // For this test, we want to send the RegisterChangeListener message after the shard
 157             // has recovered from persistence and before it becomes the leader. So we subclass
 158             // Shard to override onReceiveCommand and, when the first ElectionTimeout is received,
 159             // we know that the shard has been initialized to a follower and has started the
 160             // election process. The following 2 CountDownLatches are used to coordinate the
 161             // ElectionTimeout with the sending of the RegisterChangeListener message.
 162             final CountDownLatch onFirstElectionTimeout = new CountDownLatch(1);
 163             final CountDownLatch onChangeListenerRegistered = new CountDownLatch(1);
 164             Creator<Shard> creator = new Creator<Shard>() {
 165                 boolean firstElectionTimeout = true;
 166
 167                 @Override
 168                 public Shard create() throws Exception {
 169                     // Use a non persistent provider because this test actually invokes persist on the journal
 170                     // this will cause all other messages to not be queued properly after that.
 171                     // The basic issue is that you cannot use TestActorRef with a persistent actor (at least when
 172                     // it does do a persist)
 173                     return new Shard(shardID, Collections.<String,String>emptyMap(),
 174                             dataStoreContextBuilder.persistent(false).build(), SCHEMA_CONTEXT) {
 175                         @Override
 176                         public void onReceiveCommand(final Object message) throws Exception {
 177                             if(message instanceof ElectionTimeout && firstElectionTimeout) {
 178                                 // Got the first ElectionTimeout. We don't forward it to the
 179                                 // base Shard yet until we've sent the RegisterChangeListener
 180                                 // message. So we signal the onFirstElectionTimeout latch to tell
 181                                 // the main thread to send the RegisterChangeListener message and
 182                                 // start a thread to wait on the onChangeListenerRegistered latch,
 183                                 // which the main thread signals after it has sent the message.
 184                                 // After the onChangeListenerRegistered is triggered, we send the
 185                                 // original ElectionTimeout message to proceed with the election.
 186                                 firstElectionTimeout = false;
 187                                 final ActorRef self = getSelf();
 188                                 new Thread() {
 189                                     @Override
 190                                     public void run() {
 191                                         Uninterruptibles.awaitUninterruptibly(
 192                                                 onChangeListenerRegistered, 5, TimeUnit.SECONDS);
 193                                         self.tell(message, self);
 194                                     }
 195                                 }.start();
 196
 197                                 onFirstElectionTimeout.countDown();
 198                             } else {
 199                                 super.onReceiveCommand(message);
 200                             }
 201                         }
 202                     };
 203                 }
 204             };
 205
 206             MockDataChangeListener listener = new MockDataChangeListener(1);
 207             ActorRef dclActor = getSystem().actorOf(DataChangeListener.props(listener),
 208                     "testRegisterChangeListenerWhenNotLeaderInitially-DataChangeListener");
 209
 210             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 211                     Props.create(new DelegatingShardCreator(creator)),
 212                     "testRegisterChangeListenerWhenNotLeaderInitially");
 213
 214             // Write initial data into the in-memory store.
 215             YangInstanceIdentifier path = TestModel.TEST_PATH;
 216             writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 217
 218             // Wait until the shard receives the first ElectionTimeout message.
 219             assertEquals("Got first ElectionTimeout", true,
 220                     onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
 221
 222             // Now send the RegisterChangeListener and wait for the reply.
 223             shard.tell(new RegisterChangeListener(path, dclActor,
 224                     AsyncDataBroker.DataChangeScope.SUBTREE), getRef());
 225
 226             RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
 227                     RegisterChangeListenerReply.class);
 228             assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
 229
 230             // Sanity check - verify the shard is not the leader yet.
 231             shard.tell(new FindLeader(), getRef());
 232             FindLeaderReply findLeadeReply =
 233                     expectMsgClass(duration("5 seconds"), FindLeaderReply.class);
 234             assertNull("Expected the shard not to be the leader", findLeadeReply.getLeaderActor());
 235
 236             // Signal the onChangeListenerRegistered latch to tell the thread above to proceed
 237             // with the election process.
 238             onChangeListenerRegistered.countDown();
 239
 240             // Wait for the shard to become the leader and notify our listener with the existing
 241             // data in the store.
 242             listener.waitForChangeEvents(path);
 243
 244             dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
 245             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 246         }};
 247     }
 248
 249     @Test
 250     public void testCreateTransaction(){
 251         new ShardTestKit(getSystem()) {{
 252             ActorRef shard = getSystem().actorOf(newShardProps(), "testCreateTransaction");
 253
 254             waitUntilLeader(shard);
 255
 256             shard.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
 257
 258             shard.tell(new CreateTransaction("txn-1",
 259                     TransactionProxy.TransactionType.READ_ONLY.ordinal() ).toSerializable(), getRef());
 260
 261             CreateTransactionReply reply = expectMsgClass(duration("3 seconds"),
 262                     CreateTransactionReply.class);
 263
 264             String path = reply.getTransactionActorPath().toString();
 265             assertTrue("Unexpected transaction path " + path,
 266                     path.contains("akka://test/user/testCreateTransaction/shard-txn-1"));
 267
 268             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 269         }};
 270     }
 271
 272     @Test
 273     public void testCreateTransactionOnChain(){
 274         new ShardTestKit(getSystem()) {{
 275             final ActorRef shard = getSystem().actorOf(newShardProps(), "testCreateTransactionOnChain");
 276
 277             waitUntilLeader(shard);
 278
 279             shard.tell(new CreateTransaction("txn-1",
 280                     TransactionProxy.TransactionType.READ_ONLY.ordinal() , "foobar").toSerializable(),
 281                     getRef());
 282
 283             CreateTransactionReply reply = expectMsgClass(duration("3 seconds"),
 284                     CreateTransactionReply.class);
 285
 286             String path = reply.getTransactionActorPath().toString();
 287             assertTrue("Unexpected transaction path " + path,
 288                     path.contains("akka://test/user/testCreateTransactionOnChain/shard-txn-1"));
 289
 290             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 291         }};
 292     }
 293
 294     @SuppressWarnings("serial")
 295     @Test
 296     public void testPeerAddressResolved() throws Exception {
 297         new ShardTestKit(getSystem()) {{
 298             final CountDownLatch recoveryComplete = new CountDownLatch(1);
 299             class TestShard extends Shard {
 300                 TestShard() {
 301                     super(shardID, Collections.<String, String>singletonMap(shardID.toString(), null),
 302                             newDatastoreContext(), SCHEMA_CONTEXT);
 303                 }
 304
 305                 Map<String, String> getPeerAddresses() {
 306                     return getRaftActorContext().getPeerAddresses();
 307                 }
 308
 309                 @Override
 310                 protected void onRecoveryComplete() {
 311                     try {
 312                         super.onRecoveryComplete();
 313                     } finally {
 314                         recoveryComplete.countDown();
 315                     }
 316                 }
 317             }
 318
 319             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 320                     Props.create(new DelegatingShardCreator(new Creator<Shard>() {
 321                         @Override
 322                         public TestShard create() throws Exception {
 323                             return new TestShard();
 324                         }
 325                     })), "testPeerAddressResolved");
 326
 327             //waitUntilLeader(shard);
 328             assertEquals("Recovery complete", true,
 329                     Uninterruptibles.awaitUninterruptibly(recoveryComplete, 5, TimeUnit.SECONDS));
 330
 331             String address = "akka://foobar";
 332             shard.underlyingActor().onReceiveCommand(new PeerAddressResolved(shardID.toString(), address));
 333
 334             assertEquals("getPeerAddresses", address,
 335                     ((TestShard)shard.underlyingActor()).getPeerAddresses().get(shardID.toString()));
 336
 337             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 338         }};
 339     }
 340
 341     @Test
 342     public void testApplySnapshot() throws Exception {
 343         TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(),
 344                 "testApplySnapshot");
 345
 346         DataTree store = InMemoryDataTreeFactory.getInstance().create();
 347         store.setSchemaContext(SCHEMA_CONTEXT);
 348
 349         writeToStore(store, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 350
 351         YangInstanceIdentifier root = YangInstanceIdentifier.builder().build();
 352         NormalizedNode<?,?> expected = readStore(store, root);
 353
 354         ApplySnapshot applySnapshot = new ApplySnapshot(Snapshot.create(
 355                 SerializationUtils.serializeNormalizedNode(expected),
 356                 Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4));
 357
 358         shard.underlyingActor().onReceiveCommand(applySnapshot);
 359
 360         NormalizedNode<?,?> actual = readStore(shard, root);
 361
 362         assertEquals("Root node", expected, actual);
 363
 364         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 365     }
 366
 367     @Test
 368     public void testApplyState() throws Exception {
 369
 370         TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(), "testApplyState");
 371
 372         NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 373
 374         ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2,
 375                 newModificationPayload(new WriteModification(TestModel.TEST_PATH, node))));
 376
 377         shard.underlyingActor().onReceiveCommand(applyState);
 378
 379         NormalizedNode<?,?> actual = readStore(shard, TestModel.TEST_PATH);
 380         assertEquals("Applied state", node, actual);
 381
 382         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 383     }
 384
 385     @Test
 386     public void testApplyStateWithCandidatePayload() throws Exception {
 387
 388         TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(), "testApplyState");
 389
 390         NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 391         DataTreeCandidate candidate = DataTreeCandidates.fromNormalizedNode(TestModel.TEST_PATH, node);
 392
 393         ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2,
 394                 DataTreeCandidatePayload.create(candidate)));
 395
 396         shard.underlyingActor().onReceiveCommand(applyState);
 397
 398         NormalizedNode<?,?> actual = readStore(shard, TestModel.TEST_PATH);
 399         assertEquals("Applied state", node, actual);
 400
 401         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 402     }
 403
 404     DataTree setupInMemorySnapshotStore() throws DataValidationFailedException {
 405         DataTree testStore = InMemoryDataTreeFactory.getInstance().create();
 406         testStore.setSchemaContext(SCHEMA_CONTEXT);
 407
 408         writeToStore(testStore, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 409
 410         NormalizedNode<?, ?> root = readStore(testStore, YangInstanceIdentifier.builder().build());
 411
 412         InMemorySnapshotStore.addSnapshot(shardID.toString(), Snapshot.create(
 413                 SerializationUtils.serializeNormalizedNode(root),
 414                 Collections.<ReplicatedLogEntry>emptyList(), 0, 1, -1, -1));
 415         return testStore;
 416     }
 417
 418     private static DataTreeCandidatePayload payloadForModification(DataTree source, DataTreeModification mod) throws DataValidationFailedException {
 419         source.validate(mod);
 420         final DataTreeCandidate candidate = source.prepare(mod);
 421         source.commit(candidate);
 422         return DataTreeCandidatePayload.create(candidate);
 423     }
 424
 425     @Test
 426     public void testDataTreeCandidateRecovery() throws Exception {
 427         // Set up the InMemorySnapshotStore.
 428         final DataTree source = setupInMemorySnapshotStore();
 429
 430         final DataTreeModification writeMod = source.takeSnapshot().newModification();
 431         writeMod.write(TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
 432
 433         // Set up the InMemoryJournal.
 434         InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, payloadForModification(source, writeMod)));
 435
 436         int nListEntries = 16;
 437         Set<Integer> listEntryKeys = new HashSet<>();
 438
 439         // Add some ModificationPayload entries
 440         for (int i = 1; i <= nListEntries; i++) {
 441             listEntryKeys.add(Integer.valueOf(i));
 442
 443             YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
 444                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build();
 445
 446             final DataTreeModification mod = source.takeSnapshot().newModification();
 447             mod.merge(path, ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
 448
 449             InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1,
 450                 payloadForModification(source, mod)));
 451         }
 452
 453         InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1,
 454                 new ApplyJournalEntries(nListEntries));
 455
 456         testRecovery(listEntryKeys);
 457     }
 458
 459     @Test
 460     public void testModicationRecovery() throws Exception {
 461
 462         // Set up the InMemorySnapshotStore.
 463         setupInMemorySnapshotStore();
 464
 465         // Set up the InMemoryJournal.
 466
 467         InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, newModificationPayload(
 468                   new WriteModification(TestModel.OUTER_LIST_PATH,
 469                           ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build()))));
 470
 471         int nListEntries = 16;
 472         Set<Integer> listEntryKeys = new HashSet<>();
 473
 474         // Add some ModificationPayload entries
 475         for(int i = 1; i <= nListEntries; i++) {
 476             listEntryKeys.add(Integer.valueOf(i));
 477             YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
 478                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build();
 479             Modification mod = new MergeModification(path,
 480                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
 481             InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1,
 482                     newModificationPayload(mod)));
 483         }
 484
 485         InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1,
 486                 new ApplyJournalEntries(nListEntries));
 487
 488         testRecovery(listEntryKeys);
 489     }
 490
 491     private static ModificationPayload newModificationPayload(final Modification... mods) throws IOException {
 492         MutableCompositeModification compMod = new MutableCompositeModification();
 493         for(Modification mod: mods) {
 494             compMod.addModification(mod);
 495         }
 496
 497         return new ModificationPayload(compMod);
 498     }
 499
 500     @Test
 501     public void testConcurrentThreePhaseCommits() throws Throwable {
 502         new ShardTestKit(getSystem()) {{
 503             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 504                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
 505                     "testConcurrentThreePhaseCommits");
 506
 507             waitUntilLeader(shard);
 508
 509          // Setup 3 simulated transactions with mock cohorts backed by real cohorts.
 510
 511             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
 512
 513             String transactionID1 = "tx1";
 514             MutableCompositeModification modification1 = new MutableCompositeModification();
 515             ShardDataTreeCohort cohort1 = setupMockWriteTransaction("cohort1", dataStore,
 516                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification1);
 517
 518             String transactionID2 = "tx2";
 519             MutableCompositeModification modification2 = new MutableCompositeModification();
 520             ShardDataTreeCohort cohort2 = setupMockWriteTransaction("cohort2", dataStore,
 521                     TestModel.OUTER_LIST_PATH,
 522                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(),
 523                     modification2);
 524
 525             String transactionID3 = "tx3";
 526             MutableCompositeModification modification3 = new MutableCompositeModification();
 527             ShardDataTreeCohort cohort3 = setupMockWriteTransaction("cohort3", dataStore,
 528                     YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
 529                         .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
 530                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1),
 531                     modification3);
 532
 533             long timeoutSec = 5;
 534             final FiniteDuration duration = FiniteDuration.create(timeoutSec, TimeUnit.SECONDS);
 535             final Timeout timeout = new Timeout(duration);
 536
 537             // Simulate the ForwardedReadyTransaction message for the first Tx that would be sent
 538             // by the ShardTransaction.
 539
 540             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
 541                     cohort1, modification1, true, false), getRef());
 542             ReadyTransactionReply readyReply = ReadyTransactionReply.fromSerializable(
 543                     expectMsgClass(duration, ReadyTransactionReply.class));
 544             assertEquals("Cohort path", shard.path().toString(), readyReply.getCohortPath());
 545
 546             // Send the CanCommitTransaction message for the first Tx.
 547
 548             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
 549             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
 550                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
 551             assertEquals("Can commit", true, canCommitReply.getCanCommit());
 552
 553             // Send the ForwardedReadyTransaction for the next 2 Tx's.
 554
 555             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
 556                     cohort2, modification2, true, false), getRef());
 557             expectMsgClass(duration, ReadyTransactionReply.class);
 558
 559             shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION,
 560                     cohort3, modification3, true, false), getRef());
 561             expectMsgClass(duration, ReadyTransactionReply.class);
 562
 563             // Send the CanCommitTransaction message for the next 2 Tx's. These should get queued and
 564             // processed after the first Tx completes.
 565
 566             Future<Object> canCommitFuture1 = Patterns.ask(shard,
 567                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
 568
 569             Future<Object> canCommitFuture2 = Patterns.ask(shard,
 570                     new CanCommitTransaction(transactionID3).toSerializable(), timeout);
 571
 572             // Send the CommitTransaction message for the first Tx. After it completes, it should
 573             // trigger the 2nd Tx to proceed which should in turn then trigger the 3rd.
 574
 575             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
 576             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
 577
 578             // Wait for the next 2 Tx's to complete.
 579
 580             final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 581             final CountDownLatch commitLatch = new CountDownLatch(2);
 582
 583             class OnFutureComplete extends OnComplete<Object> {
 584                 private final Class<?> expRespType;
 585
 586                 OnFutureComplete(final Class<?> expRespType) {
 587                     this.expRespType = expRespType;
 588                 }
 589
 590                 @Override
 591                 public void onComplete(final Throwable error, final Object resp) {
 592                     if(error != null) {
 593                         caughtEx.set(new AssertionError(getClass().getSimpleName() + " failure", error));
 594                     } else {
 595                         try {
 596                             assertEquals("Commit response type", expRespType, resp.getClass());
 597                             onSuccess(resp);
 598                         } catch (Exception e) {
 599                             caughtEx.set(e);
 600                         }
 601                     }
 602                 }
 603
 604                 void onSuccess(final Object resp) throws Exception {
 605                 }
 606             }
 607
 608             class OnCommitFutureComplete extends OnFutureComplete {
 609                 OnCommitFutureComplete() {
 610                     super(CommitTransactionReply.SERIALIZABLE_CLASS);
 611                 }
 612
 613                 @Override
 614                 public void onComplete(final Throwable error, final Object resp) {
 615                     super.onComplete(error, resp);
 616                     commitLatch.countDown();
 617                 }
 618             }
 619
 620             class OnCanCommitFutureComplete extends OnFutureComplete {
 621                 private final String transactionID;
 622
 623                 OnCanCommitFutureComplete(final String transactionID) {
 624                     super(CanCommitTransactionReply.SERIALIZABLE_CLASS);
 625                     this.transactionID = transactionID;
 626                 }
 627
 628                 @Override
 629                 void onSuccess(final Object resp) throws Exception {
 630                     CanCommitTransactionReply canCommitReply =
 631                             CanCommitTransactionReply.fromSerializable(resp);
 632                     assertEquals("Can commit", true, canCommitReply.getCanCommit());
 633
 634                     Future<Object> commitFuture = Patterns.ask(shard,
 635                             new CommitTransaction(transactionID).toSerializable(), timeout);
 636                     commitFuture.onComplete(new OnCommitFutureComplete(), getSystem().dispatcher());
 637                 }
 638             }
 639
 640             canCommitFuture1.onComplete(new OnCanCommitFutureComplete(transactionID2),
 641                     getSystem().dispatcher());
 642
 643             canCommitFuture2.onComplete(new OnCanCommitFutureComplete(transactionID3),
 644                     getSystem().dispatcher());
 645
 646             boolean done = commitLatch.await(timeoutSec, TimeUnit.SECONDS);
 647
 648             if(caughtEx.get() != null) {
 649                 throw caughtEx.get();
 650             }
 651
 652             assertEquals("Commits complete", true, done);
 653
 654             InOrder inOrder = inOrder(cohort1, cohort2, cohort3);
 655             inOrder.verify(cohort1).canCommit();
 656             inOrder.verify(cohort1).preCommit();
 657             inOrder.verify(cohort1).commit();
 658             inOrder.verify(cohort2).canCommit();
 659             inOrder.verify(cohort2).preCommit();
 660             inOrder.verify(cohort2).commit();
 661             inOrder.verify(cohort3).canCommit();
 662             inOrder.verify(cohort3).preCommit();
 663             inOrder.verify(cohort3).commit();
 664
 665             // Verify data in the data store.
 666
 667             verifyOuterListEntry(shard, 1);
 668
 669             verifyLastApplied(shard, 2);
 670
 671             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 672         }};
 673     }
 674
 675     private static BatchedModifications newBatchedModifications(String transactionID, YangInstanceIdentifier path,
 676             NormalizedNode<?, ?> data, boolean ready, boolean doCommitOnReady) {
 677         return newBatchedModifications(transactionID, null, path, data, ready, doCommitOnReady);
 678     }
 679
 680     private static BatchedModifications newBatchedModifications(String transactionID, String transactionChainID,
 681             YangInstanceIdentifier path, NormalizedNode<?, ?> data, boolean ready, boolean doCommitOnReady) {
 682         BatchedModifications batched = new BatchedModifications(transactionID, CURRENT_VERSION, transactionChainID);
 683         batched.addModification(new WriteModification(path, data));
 684         batched.setReady(ready);
 685         batched.setDoCommitOnReady(doCommitOnReady);
 686         return batched;
 687     }
 688
 689     @Test
 690     public void testBatchedModificationsWithNoCommitOnReady() throws Throwable {
 691         new ShardTestKit(getSystem()) {{
 692             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 693                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
 694                     "testBatchedModificationsWithNoCommitOnReady");
 695
 696             waitUntilLeader(shard);
 697
 698             final String transactionID = "tx";
 699             FiniteDuration duration = duration("5 seconds");
 700
 701             final AtomicReference<ShardDataTreeCohort> mockCohort = new AtomicReference<>();
 702             ShardCommitCoordinator.CohortDecorator cohortDecorator = new ShardCommitCoordinator.CohortDecorator() {
 703                 @Override
 704                 public ShardDataTreeCohort decorate(String txID, ShardDataTreeCohort actual) {
 705                     if(mockCohort.get() == null) {
 706                         mockCohort.set(createDelegatingMockCohort("cohort", actual));
 707                     }
 708
 709                     return mockCohort.get();
 710                 }
 711             };
 712
 713             shard.underlyingActor().getCommitCoordinator().setCohortDecorator(cohortDecorator);
 714
 715             // Send a BatchedModifications to start a transaction.
 716
 717             shard.tell(newBatchedModifications(transactionID, TestModel.TEST_PATH,
 718                     ImmutableNodes.containerNode(TestModel.TEST_QNAME), false, false), getRef());
 719             expectMsgClass(duration, BatchedModificationsReply.class);
 720
 721             // Send a couple more BatchedModifications.
 722
 723             shard.tell(newBatchedModifications(transactionID, TestModel.OUTER_LIST_PATH,
 724                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false), getRef());
 725             expectMsgClass(duration, BatchedModificationsReply.class);
 726
 727             shard.tell(newBatchedModifications(transactionID, YangInstanceIdentifier.builder(
 728                     TestModel.OUTER_LIST_PATH).nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
 729                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1), true, false), getRef());
 730             expectMsgClass(duration, ReadyTransactionReply.class);
 731
 732             // Send the CanCommitTransaction message.
 733
 734             shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
 735             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
 736                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
 737             assertEquals("Can commit", true, canCommitReply.getCanCommit());
 738
 739             // Send the CanCommitTransaction message.
 740
 741             shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
 742             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
 743
 744             InOrder inOrder = inOrder(mockCohort.get());
 745             inOrder.verify(mockCohort.get()).canCommit();
 746             inOrder.verify(mockCohort.get()).preCommit();
 747             inOrder.verify(mockCohort.get()).commit();
 748
 749             // Verify data in the data store.
 750
 751             verifyOuterListEntry(shard, 1);
 752
 753             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 754         }};
 755     }
 756
 757     @Test
 758     public void testBatchedModificationsWithCommitOnReady() throws Throwable {
 759         new ShardTestKit(getSystem()) {{
 760             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 761                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
 762                     "testBatchedModificationsWithCommitOnReady");
 763
 764             waitUntilLeader(shard);
 765
 766             final String transactionID = "tx";
 767             FiniteDuration duration = duration("5 seconds");
 768
 769             final AtomicReference<ShardDataTreeCohort> mockCohort = new AtomicReference<>();
 770             ShardCommitCoordinator.CohortDecorator cohortDecorator = new ShardCommitCoordinator.CohortDecorator() {
 771                 @Override
 772                 public ShardDataTreeCohort decorate(String txID, ShardDataTreeCohort actual) {
 773                     if(mockCohort.get() == null) {
 774                         mockCohort.set(createDelegatingMockCohort("cohort", actual));
 775                     }
 776
 777                     return mockCohort.get();
 778                 }
 779             };
 780
 781             shard.underlyingActor().getCommitCoordinator().setCohortDecorator(cohortDecorator);
 782
 783             // Send a BatchedModifications to start a transaction.
 784
 785             shard.tell(newBatchedModifications(transactionID, TestModel.TEST_PATH,
 786                     ImmutableNodes.containerNode(TestModel.TEST_QNAME), false, false), getRef());
 787             expectMsgClass(duration, BatchedModificationsReply.class);
 788
 789             // Send a couple more BatchedModifications.
 790
 791             shard.tell(newBatchedModifications(transactionID, TestModel.OUTER_LIST_PATH,
 792                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false), getRef());
 793             expectMsgClass(duration, BatchedModificationsReply.class);
 794
 795             shard.tell(newBatchedModifications(transactionID, YangInstanceIdentifier.builder(
 796                     TestModel.OUTER_LIST_PATH).nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
 797                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1), true, true), getRef());
 798
 799             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
 800
 801             InOrder inOrder = inOrder(mockCohort.get());
 802             inOrder.verify(mockCohort.get()).canCommit();
 803             inOrder.verify(mockCohort.get()).preCommit();
 804             inOrder.verify(mockCohort.get()).commit();
 805
 806             // Verify data in the data store.
 807
 808             verifyOuterListEntry(shard, 1);
 809
 810             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 811         }};
 812     }
 813
 814     @SuppressWarnings("unchecked")
 815     private static void verifyOuterListEntry(final TestActorRef<Shard> shard, Object expIDValue) throws Exception {
 816         NormalizedNode<?, ?> outerList = readStore(shard, TestModel.OUTER_LIST_PATH);
 817         assertNotNull(TestModel.OUTER_LIST_QNAME.getLocalName() + " not found", outerList);
 818         assertTrue(TestModel.OUTER_LIST_QNAME.getLocalName() + " value is not Iterable",
 819                 outerList.getValue() instanceof Iterable);
 820         Object entry = ((Iterable<Object>)outerList.getValue()).iterator().next();
 821         assertTrue(TestModel.OUTER_LIST_QNAME.getLocalName() + " entry is not MapEntryNode",
 822                 entry instanceof MapEntryNode);
 823         MapEntryNode mapEntry = (MapEntryNode)entry;
 824         Optional<DataContainerChild<? extends PathArgument, ?>> idLeaf =
 825                 mapEntry.getChild(new YangInstanceIdentifier.NodeIdentifier(TestModel.ID_QNAME));
 826         assertTrue("Missing leaf " + TestModel.ID_QNAME.getLocalName(), idLeaf.isPresent());
 827         assertEquals(TestModel.ID_QNAME.getLocalName() + " value", expIDValue, idLeaf.get().getValue());
 828     }
 829
 830     @Test
 831     public void testBatchedModificationsOnTransactionChain() throws Throwable {
 832         new ShardTestKit(getSystem()) {{
 833             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 834                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
 835                     "testBatchedModificationsOnTransactionChain");
 836
 837             waitUntilLeader(shard);
 838
 839             String transactionChainID = "txChain";
 840             String transactionID1 = "tx1";
 841             String transactionID2 = "tx2";
 842
 843             FiniteDuration duration = duration("5 seconds");
 844
 845             // Send a BatchedModifications to start a chained write transaction and ready it.
 846
 847             ContainerNode containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 848             YangInstanceIdentifier path = TestModel.TEST_PATH;
 849             shard.tell(newBatchedModifications(transactionID1, transactionChainID, path,
 850                     containerNode, true, false), getRef());
 851             expectMsgClass(duration, ReadyTransactionReply.class);
 852
 853             // Create a read Tx on the same chain.
 854
 855             shard.tell(new CreateTransaction(transactionID2, TransactionProxy.TransactionType.READ_ONLY.ordinal() ,
 856                     transactionChainID).toSerializable(), getRef());
 857
 858             CreateTransactionReply createReply = expectMsgClass(duration("3 seconds"), CreateTransactionReply.class);
 859
 860             getSystem().actorSelection(createReply.getTransactionActorPath()).tell(new ReadData(path), getRef());
 861             ReadDataReply readReply = expectMsgClass(duration("3 seconds"), ReadDataReply.class);
 862             assertEquals("Read node", containerNode, readReply.getNormalizedNode());
 863
 864             // Commit the write transaction.
 865
 866             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
 867             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
 868                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
 869             assertEquals("Can commit", true, canCommitReply.getCanCommit());
 870
 871             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
 872             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
 873
 874             // Verify data in the data store.
 875
 876             NormalizedNode<?, ?> actualNode = readStore(shard, path);
 877             assertEquals("Stored node", containerNode, actualNode);
 878
 879             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 880         }};
 881     }
 882
 883     @Test
 884     public void testOnBatchedModificationsWhenNotLeader() {
 885         final AtomicBoolean overrideLeaderCalls = new AtomicBoolean();
 886         new ShardTestKit(getSystem()) {{
 887             Creator<Shard> creator = new Creator<Shard>() {
 888                 private static final long serialVersionUID = 1L;
 889
 890                 @Override
 891                 public Shard create() throws Exception {
 892                     return new Shard(shardID, Collections.<String,String>emptyMap(),
 893                             newDatastoreContext(), SCHEMA_CONTEXT) {
 894                         @Override
 895                         protected boolean isLeader() {
 896                             return overrideLeaderCalls.get() ? false : super.isLeader();
 897                         }
 898
 899                         @Override
 900                         protected ActorSelection getLeader() {
 901                             return overrideLeaderCalls.get() ? getSystem().actorSelection(getRef().path()) :
 902                                 super.getLeader();
 903                         }
 904                     };
 905                 }
 906             };
 907
 908             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 909                     Props.create(new DelegatingShardCreator(creator)), "testOnBatchedModificationsWhenNotLeader");
 910
 911             waitUntilLeader(shard);
 912
 913             overrideLeaderCalls.set(true);
 914
 915             BatchedModifications batched = new BatchedModifications("tx", DataStoreVersions.CURRENT_VERSION, "");
 916
 917             shard.tell(batched, ActorRef.noSender());
 918
 919             expectMsgEquals(batched);
 920
 921             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 922         }};
 923     }
 924
 925     @Test
 926     public void testForwardedReadyTransactionWithImmediateCommit() throws Exception{
 927         new ShardTestKit(getSystem()) {{
 928             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 929                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
 930                     "testForwardedReadyTransactionWithImmediateCommit");
 931
 932             waitUntilLeader(shard);
 933
 934             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
 935
 936             String transactionID = "tx1";
 937             MutableCompositeModification modification = new MutableCompositeModification();
 938             NormalizedNode<?, ?> containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 939             ShardDataTreeCohort cohort = setupMockWriteTransaction("cohort", dataStore,
 940                     TestModel.TEST_PATH, containerNode, modification);
 941
 942             FiniteDuration duration = duration("5 seconds");
 943
 944             // Simulate the ForwardedReadyTransaction messages that would be sent
 945             // by the ShardTransaction.
 946
 947             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
 948                     cohort, modification, true, true), getRef());
 949
 950             expectMsgClass(duration, ThreePhaseCommitCohortMessages.CommitTransactionReply.class);
 951
 952             InOrder inOrder = inOrder(cohort);
 953             inOrder.verify(cohort).canCommit();
 954             inOrder.verify(cohort).preCommit();
 955             inOrder.verify(cohort).commit();
 956
 957             NormalizedNode<?, ?> actualNode = readStore(shard, TestModel.TEST_PATH);
 958             assertEquals(TestModel.TEST_QNAME.getLocalName(), containerNode, actualNode);
 959
 960             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
 961         }};
 962     }
 963
 964     @Test
 965     public void testCommitWithPersistenceDisabled() throws Throwable {
 966         dataStoreContextBuilder.persistent(false);
 967         new ShardTestKit(getSystem()) {{
 968             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
 969                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
 970                     "testCommitWithPersistenceDisabled");
 971
 972             waitUntilLeader(shard);
 973
 974             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
 975
 976             // Setup a simulated transactions with a mock cohort.
 977
 978             String transactionID = "tx";
 979             MutableCompositeModification modification = new MutableCompositeModification();
 980             NormalizedNode<?, ?> containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 981             ShardDataTreeCohort cohort = setupMockWriteTransaction("cohort", dataStore,
 982                     TestModel.TEST_PATH, containerNode, modification);
 983
 984             FiniteDuration duration = duration("5 seconds");
 985
 986             // Simulate the ForwardedReadyTransaction messages that would be sent
 987             // by the ShardTransaction.
 988
 989             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
 990                     cohort, modification, true, false), getRef());
 991             expectMsgClass(duration, ReadyTransactionReply.class);
 992
 993             // Send the CanCommitTransaction message.
 994
 995             shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
 996             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
 997                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
 998             assertEquals("Can commit", true, canCommitReply.getCanCommit());
 999
1000             // Send the CanCommitTransaction message.
1001
1002             shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1003             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1004
1005             InOrder inOrder = inOrder(cohort);
1006             inOrder.verify(cohort).canCommit();
1007             inOrder.verify(cohort).preCommit();
1008             inOrder.verify(cohort).commit();
1009
1010             NormalizedNode<?, ?> actualNode = readStore(shard, TestModel.TEST_PATH);
1011             assertEquals(TestModel.TEST_QNAME.getLocalName(), containerNode, actualNode);
1012
1013             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1014         }};
1015     }
1016
1017     private static DataTreeCandidateTip mockCandidate(final String name) {
1018         DataTreeCandidateTip mockCandidate = mock(DataTreeCandidateTip.class, name);
1019         DataTreeCandidateNode mockCandidateNode = mock(DataTreeCandidateNode.class, name + "-node");
1020         doReturn(ModificationType.WRITE).when(mockCandidateNode).getModificationType();
1021         doReturn(Optional.of(ImmutableNodes.containerNode(CARS_QNAME))).when(mockCandidateNode).getDataAfter();
1022         doReturn(YangInstanceIdentifier.builder().build()).when(mockCandidate).getRootPath();
1023         doReturn(mockCandidateNode).when(mockCandidate).getRootNode();
1024         return mockCandidate;
1025     }
1026
1027     private static DataTreeCandidateTip mockUnmodifiedCandidate(final String name) {
1028         DataTreeCandidateTip mockCandidate = mock(DataTreeCandidateTip.class, name);
1029         DataTreeCandidateNode mockCandidateNode = mock(DataTreeCandidateNode.class, name + "-node");
1030         doReturn(ModificationType.UNMODIFIED).when(mockCandidateNode).getModificationType();
1031         doReturn(YangInstanceIdentifier.builder().build()).when(mockCandidate).getRootPath();
1032         doReturn(mockCandidateNode).when(mockCandidate).getRootNode();
1033         return mockCandidate;
1034     }
1035
1036     @Test
1037     public void testCommitWhenTransactionHasNoModifications(){
1038         // Note that persistence is enabled which would normally result in the entry getting written to the journal
1039         // but here that need not happen
1040         new ShardTestKit(getSystem()) {
1041             {
1042                 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1043                         newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1044                         "testCommitWhenTransactionHasNoModifications");
1045
1046                 waitUntilLeader(shard);
1047
1048                 String transactionID = "tx1";
1049                 MutableCompositeModification modification = new MutableCompositeModification();
1050                 ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1051                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1052                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).preCommit();
1053                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).commit();
1054                 doReturn(mockUnmodifiedCandidate("cohort1-candidate")).when(cohort).getCandidate();
1055
1056                 FiniteDuration duration = duration("5 seconds");
1057
1058                 // Simulate the ForwardedReadyTransaction messages that would be sent
1059                 // by the ShardTransaction.
1060
1061                 shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1062                         cohort, modification, true, false), getRef());
1063                 expectMsgClass(duration, ReadyTransactionReply.class);
1064
1065                 // Send the CanCommitTransaction message.
1066
1067                 shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
1068                 CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1069                         expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1070                 assertEquals("Can commit", true, canCommitReply.getCanCommit());
1071
1072                 shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1073                 expectMsgClass(duration, ThreePhaseCommitCohortMessages.CommitTransactionReply.class);
1074
1075                 InOrder inOrder = inOrder(cohort);
1076                 inOrder.verify(cohort).canCommit();
1077                 inOrder.verify(cohort).preCommit();
1078                 inOrder.verify(cohort).commit();
1079
1080                 // Use MBean for verification
1081                 // Committed transaction count should increase as usual
1082                 assertEquals(1,shard.underlyingActor().getShardMBean().getCommittedTransactionsCount());
1083
1084                 // Commit index should not advance because this does not go into the journal
1085                 assertEquals(-1, shard.underlyingActor().getShardMBean().getCommitIndex());
1086
1087                 shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1088
1089             }
1090         };
1091     }
1092
1093     @Test
1094     public void testCommitWhenTransactionHasModifications(){
1095         new ShardTestKit(getSystem()) {
1096             {
1097                 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1098                         newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1099                         "testCommitWhenTransactionHasModifications");
1100
1101                 waitUntilLeader(shard);
1102
1103                 String transactionID = "tx1";
1104                 MutableCompositeModification modification = new MutableCompositeModification();
1105                 modification.addModification(new DeleteModification(YangInstanceIdentifier.builder().build()));
1106                 ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1107                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1108                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).preCommit();
1109                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).commit();
1110                 doReturn(mockCandidate("cohort1-candidate")).when(cohort).getCandidate();
1111
1112                 FiniteDuration duration = duration("5 seconds");
1113
1114                 // Simulate the ForwardedReadyTransaction messages that would be sent
1115                 // by the ShardTransaction.
1116
1117                 shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1118                         cohort, modification, true, false), getRef());
1119                 expectMsgClass(duration, ReadyTransactionReply.class);
1120
1121                 // Send the CanCommitTransaction message.
1122
1123                 shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
1124                 CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1125                         expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1126                 assertEquals("Can commit", true, canCommitReply.getCanCommit());
1127
1128                 shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1129                 expectMsgClass(duration, ThreePhaseCommitCohortMessages.CommitTransactionReply.class);
1130
1131                 InOrder inOrder = inOrder(cohort);
1132                 inOrder.verify(cohort).canCommit();
1133                 inOrder.verify(cohort).preCommit();
1134                 inOrder.verify(cohort).commit();
1135
1136                 // Use MBean for verification
1137                 // Committed transaction count should increase as usual
1138                 assertEquals(1, shard.underlyingActor().getShardMBean().getCommittedTransactionsCount());
1139
1140                 // Commit index should advance as we do not have an empty modification
1141                 assertEquals(0, shard.underlyingActor().getShardMBean().getCommitIndex());
1142
1143                 shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1144
1145             }
1146         };
1147     }
1148
1149     @Test
1150     public void testCommitPhaseFailure() throws Throwable {
1151         new ShardTestKit(getSystem()) {{
1152             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1153                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1154                     "testCommitPhaseFailure");
1155
1156             waitUntilLeader(shard);
1157
1158             // Setup 2 simulated transactions with mock cohorts. The first one fails in the
1159             // commit phase.
1160
1161             String transactionID1 = "tx1";
1162             MutableCompositeModification modification1 = new MutableCompositeModification();
1163             ShardDataTreeCohort cohort1 = mock(ShardDataTreeCohort.class, "cohort1");
1164             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort1).canCommit();
1165             doReturn(Futures.immediateFuture(null)).when(cohort1).preCommit();
1166             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort1).commit();
1167             doReturn(mockCandidate("cohort1-candidate")).when(cohort1).getCandidate();
1168
1169             String transactionID2 = "tx2";
1170             MutableCompositeModification modification2 = new MutableCompositeModification();
1171             ShardDataTreeCohort cohort2 = mock(ShardDataTreeCohort.class, "cohort2");
1172             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort2).canCommit();
1173
1174             FiniteDuration duration = duration("5 seconds");
1175             final Timeout timeout = new Timeout(duration);
1176
1177             // Simulate the ForwardedReadyTransaction messages that would be sent
1178             // by the ShardTransaction.
1179
1180             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1181                     cohort1, modification1, true, false), getRef());
1182             expectMsgClass(duration, ReadyTransactionReply.class);
1183
1184             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1185                     cohort2, modification2, true, false), getRef());
1186             expectMsgClass(duration, ReadyTransactionReply.class);
1187
1188             // Send the CanCommitTransaction message for the first Tx.
1189
1190             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1191             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1192                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1193             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1194
1195             // Send the CanCommitTransaction message for the 2nd Tx. This should get queued and
1196             // processed after the first Tx completes.
1197
1198             Future<Object> canCommitFuture = Patterns.ask(shard,
1199                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
1200
1201             // Send the CommitTransaction message for the first Tx. This should send back an error
1202             // and trigger the 2nd Tx to proceed.
1203
1204             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
1205             expectMsgClass(duration, akka.actor.Status.Failure.class);
1206
1207             // Wait for the 2nd Tx to complete the canCommit phase.
1208
1209             final CountDownLatch latch = new CountDownLatch(1);
1210             canCommitFuture.onComplete(new OnComplete<Object>() {
1211                 @Override
1212                 public void onComplete(final Throwable t, final Object resp) {
1213                     latch.countDown();
1214                 }
1215             }, getSystem().dispatcher());
1216
1217             assertEquals("2nd CanCommit complete", true, latch.await(5, TimeUnit.SECONDS));
1218
1219             InOrder inOrder = inOrder(cohort1, cohort2);
1220             inOrder.verify(cohort1).canCommit();
1221             inOrder.verify(cohort1).preCommit();
1222             inOrder.verify(cohort1).commit();
1223             inOrder.verify(cohort2).canCommit();
1224
1225             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1226         }};
1227     }
1228
1229     @Test
1230     public void testPreCommitPhaseFailure() throws Throwable {
1231         new ShardTestKit(getSystem()) {{
1232             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1233                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1234                     "testPreCommitPhaseFailure");
1235
1236             waitUntilLeader(shard);
1237
1238             String transactionID1 = "tx1";
1239             MutableCompositeModification modification1 = new MutableCompositeModification();
1240             ShardDataTreeCohort cohort1 = mock(ShardDataTreeCohort.class, "cohort1");
1241             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort1).canCommit();
1242             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort1).preCommit();
1243
1244             String transactionID2 = "tx2";
1245             MutableCompositeModification modification2 = new MutableCompositeModification();
1246             ShardDataTreeCohort cohort2 = mock(ShardDataTreeCohort.class, "cohort2");
1247             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort2).canCommit();
1248
1249             FiniteDuration duration = duration("5 seconds");
1250             final Timeout timeout = new Timeout(duration);
1251
1252             // Simulate the ForwardedReadyTransaction messages that would be sent
1253             // by the ShardTransaction.
1254
1255             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1256                     cohort1, modification1, true, false), getRef());
1257             expectMsgClass(duration, ReadyTransactionReply.class);
1258
1259             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1260                     cohort2, modification2, true, false), getRef());
1261             expectMsgClass(duration, ReadyTransactionReply.class);
1262
1263             // Send the CanCommitTransaction message for the first Tx.
1264
1265             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1266             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1267                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1268             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1269
1270             // Send the CanCommitTransaction message for the 2nd Tx. This should get queued and
1271             // processed after the first Tx completes.
1272
1273             Future<Object> canCommitFuture = Patterns.ask(shard,
1274                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
1275
1276             // Send the CommitTransaction message for the first Tx. This should send back an error
1277             // and trigger the 2nd Tx to proceed.
1278
1279             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
1280             expectMsgClass(duration, akka.actor.Status.Failure.class);
1281
1282             // Wait for the 2nd Tx to complete the canCommit phase.
1283
1284             final CountDownLatch latch = new CountDownLatch(1);
1285             canCommitFuture.onComplete(new OnComplete<Object>() {
1286                 @Override
1287                 public void onComplete(final Throwable t, final Object resp) {
1288                     latch.countDown();
1289                 }
1290             }, getSystem().dispatcher());
1291
1292             assertEquals("2nd CanCommit complete", true, latch.await(5, TimeUnit.SECONDS));
1293
1294             InOrder inOrder = inOrder(cohort1, cohort2);
1295             inOrder.verify(cohort1).canCommit();
1296             inOrder.verify(cohort1).preCommit();
1297             inOrder.verify(cohort2).canCommit();
1298
1299             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1300         }};
1301     }
1302
1303     @Test
1304     public void testCanCommitPhaseFailure() throws Throwable {
1305         new ShardTestKit(getSystem()) {{
1306             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1307                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1308                     "testCanCommitPhaseFailure");
1309
1310             waitUntilLeader(shard);
1311
1312             final FiniteDuration duration = duration("5 seconds");
1313
1314             String transactionID1 = "tx1";
1315             MutableCompositeModification modification = new MutableCompositeModification();
1316             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1317             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort).canCommit();
1318
1319             // Simulate the ForwardedReadyTransaction messages that would be sent
1320             // by the ShardTransaction.
1321
1322             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1323                     cohort, modification, true, false), getRef());
1324             expectMsgClass(duration, ReadyTransactionReply.class);
1325
1326             // Send the CanCommitTransaction message.
1327
1328             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1329             expectMsgClass(duration, akka.actor.Status.Failure.class);
1330
1331             // Send another can commit to ensure the failed one got cleaned up.
1332
1333             reset(cohort);
1334
1335             String transactionID2 = "tx2";
1336             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1337
1338             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1339                     cohort, modification, true, false), getRef());
1340             expectMsgClass(duration, ReadyTransactionReply.class);
1341
1342             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1343             CanCommitTransactionReply reply = CanCommitTransactionReply.fromSerializable(
1344                     expectMsgClass(CanCommitTransactionReply.SERIALIZABLE_CLASS));
1345             assertEquals("getCanCommit", true, reply.getCanCommit());
1346
1347             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1348         }};
1349     }
1350
1351     @Test
1352     public void testCanCommitPhaseFalseResponse() throws Throwable {
1353         new ShardTestKit(getSystem()) {{
1354             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1355                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1356                     "testCanCommitPhaseFalseResponse");
1357
1358             waitUntilLeader(shard);
1359
1360             final FiniteDuration duration = duration("5 seconds");
1361
1362             String transactionID1 = "tx1";
1363             MutableCompositeModification modification = new MutableCompositeModification();
1364             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1365             doReturn(Futures.immediateFuture(Boolean.FALSE)).when(cohort).canCommit();
1366
1367             // Simulate the ForwardedReadyTransaction messages that would be sent
1368             // by the ShardTransaction.
1369
1370             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1371                     cohort, modification, true, false), getRef());
1372             expectMsgClass(duration, ReadyTransactionReply.class);
1373
1374             // Send the CanCommitTransaction message.
1375
1376             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1377             CanCommitTransactionReply reply = CanCommitTransactionReply.fromSerializable(
1378                     expectMsgClass(CanCommitTransactionReply.SERIALIZABLE_CLASS));
1379             assertEquals("getCanCommit", false, reply.getCanCommit());
1380
1381             // Send another can commit to ensure the failed one got cleaned up.
1382
1383             reset(cohort);
1384
1385             String transactionID2 = "tx2";
1386             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1387
1388             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1389                     cohort, modification, true, false), getRef());
1390             expectMsgClass(duration, ReadyTransactionReply.class);
1391
1392             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1393             reply = CanCommitTransactionReply.fromSerializable(
1394                     expectMsgClass(CanCommitTransactionReply.SERIALIZABLE_CLASS));
1395             assertEquals("getCanCommit", true, reply.getCanCommit());
1396
1397             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1398         }};
1399     }
1400
1401     @Test
1402     public void testImmediateCommitWithCanCommitPhaseFailure() throws Throwable {
1403         new ShardTestKit(getSystem()) {{
1404             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1405                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1406                     "testImmediateCommitWithCanCommitPhaseFailure");
1407
1408             waitUntilLeader(shard);
1409
1410             final FiniteDuration duration = duration("5 seconds");
1411
1412             String transactionID1 = "tx1";
1413             MutableCompositeModification modification = new MutableCompositeModification();
1414             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1415             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort).canCommit();
1416
1417             // Simulate the ForwardedReadyTransaction messages that would be sent
1418             // by the ShardTransaction.
1419
1420             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1421                     cohort, modification, true, true), getRef());
1422
1423             expectMsgClass(duration, akka.actor.Status.Failure.class);
1424
1425             // Send another can commit to ensure the failed one got cleaned up.
1426
1427             reset(cohort);
1428
1429             String transactionID2 = "tx2";
1430             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1431             doReturn(Futures.immediateFuture(null)).when(cohort).preCommit();
1432             doReturn(Futures.immediateFuture(null)).when(cohort).commit();
1433             DataTreeCandidateTip candidate = mock(DataTreeCandidateTip.class);
1434             DataTreeCandidateNode candidateRoot = mock(DataTreeCandidateNode.class);
1435             doReturn(ModificationType.UNMODIFIED).when(candidateRoot).getModificationType();
1436             doReturn(candidateRoot).when(candidate).getRootNode();
1437             doReturn(candidate).when(cohort).getCandidate();
1438
1439             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1440                     cohort, modification, true, true), getRef());
1441
1442             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1443
1444             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1445         }};
1446     }
1447
1448     @Test
1449     public void testImmediateCommitWithCanCommitPhaseFalseResponse() throws Throwable {
1450         new ShardTestKit(getSystem()) {{
1451             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1452                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1453                     "testImmediateCommitWithCanCommitPhaseFalseResponse");
1454
1455             waitUntilLeader(shard);
1456
1457             final FiniteDuration duration = duration("5 seconds");
1458
1459             String transactionID = "tx1";
1460             MutableCompositeModification modification = new MutableCompositeModification();
1461             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1462             doReturn(Futures.immediateFuture(Boolean.FALSE)).when(cohort).canCommit();
1463
1464             // Simulate the ForwardedReadyTransaction messages that would be sent
1465             // by the ShardTransaction.
1466
1467             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1468                     cohort, modification, true, true), getRef());
1469
1470             expectMsgClass(duration, akka.actor.Status.Failure.class);
1471
1472             // Send another can commit to ensure the failed one got cleaned up.
1473
1474             reset(cohort);
1475
1476             String transactionID2 = "tx2";
1477             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1478             doReturn(Futures.immediateFuture(null)).when(cohort).preCommit();
1479             doReturn(Futures.immediateFuture(null)).when(cohort).commit();
1480             DataTreeCandidateTip candidate = mock(DataTreeCandidateTip.class);
1481             DataTreeCandidateNode candidateRoot = mock(DataTreeCandidateNode.class);
1482             doReturn(ModificationType.UNMODIFIED).when(candidateRoot).getModificationType();
1483             doReturn(candidateRoot).when(candidate).getRootNode();
1484             doReturn(candidate).when(cohort).getCandidate();
1485
1486             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1487                     cohort, modification, true, true), getRef());
1488
1489             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1490
1491             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1492         }};
1493     }
1494
1495     @Test
1496     public void testAbortBeforeFinishCommit() throws Throwable {
1497         new ShardTestKit(getSystem()) {{
1498             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1499                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1500                     "testAbortBeforeFinishCommit");
1501
1502             waitUntilLeader(shard);
1503
1504             final FiniteDuration duration = duration("5 seconds");
1505             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
1506
1507             final String transactionID = "tx1";
1508             Function<ShardDataTreeCohort, ListenableFuture<Void>> preCommit =
1509                           new Function<ShardDataTreeCohort, ListenableFuture<Void>>() {
1510                 @Override
1511                 public ListenableFuture<Void> apply(final ShardDataTreeCohort cohort) {
1512                     ListenableFuture<Void> preCommitFuture = cohort.preCommit();
1513
1514                     // Simulate an AbortTransaction message occurring during replication, after
1515                     // persisting and before finishing the commit to the in-memory store.
1516                     // We have no followers so due to optimizations in the RaftActor, it does not
1517                     // attempt replication and thus we can't send an AbortTransaction message b/c
1518                     // it would be processed too late after CommitTransaction completes. So we'll
1519                     // simulate an AbortTransaction message occurring during replication by calling
1520                     // the shard directly.
1521                     //
1522                     shard.underlyingActor().doAbortTransaction(transactionID, null);
1523
1524                     return preCommitFuture;
1525                 }
1526             };
1527
1528             MutableCompositeModification modification = new MutableCompositeModification();
1529             ShardDataTreeCohort cohort = setupMockWriteTransaction("cohort1", dataStore,
1530                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME),
1531                     modification, preCommit);
1532
1533             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1534                     cohort, modification, true, false), getRef());
1535             expectMsgClass(duration, ReadyTransactionReply.class);
1536
1537             shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
1538             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1539                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1540             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1541
1542             shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1543             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1544
1545             NormalizedNode<?, ?> node = readStore(shard, TestModel.TEST_PATH);
1546
1547             // Since we're simulating an abort occurring during replication and before finish commit,
1548             // the data should still get written to the in-memory store since we've gotten past
1549             // canCommit and preCommit and persisted the data.
1550             assertNotNull(TestModel.TEST_QNAME.getLocalName() + " not found", node);
1551
1552             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1553         }};
1554     }
1555
1556     @Test
1557     public void testTransactionCommitTimeout() throws Throwable {
1558         dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1);
1559
1560         new ShardTestKit(getSystem()) {{
1561             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1562                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1563                     "testTransactionCommitTimeout");
1564
1565             waitUntilLeader(shard);
1566
1567             final FiniteDuration duration = duration("5 seconds");
1568
1569             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
1570
1571             writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1572             writeToStore(shard, TestModel.OUTER_LIST_PATH,
1573                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
1574
1575             // Create 1st Tx - will timeout
1576
1577             String transactionID1 = "tx1";
1578             MutableCompositeModification modification1 = new MutableCompositeModification();
1579             ShardDataTreeCohort cohort1 = setupMockWriteTransaction("cohort1", dataStore,
1580                     YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1581                         .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
1582                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1),
1583                     modification1);
1584
1585             // Create 2nd Tx
1586
1587             String transactionID2 = "tx3";
1588             MutableCompositeModification modification2 = new MutableCompositeModification();
1589             YangInstanceIdentifier listNodePath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1590                 .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build();
1591             ShardDataTreeCohort cohort2 = setupMockWriteTransaction("cohort3", dataStore,
1592                     listNodePath,
1593                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2),
1594                     modification2);
1595
1596             // Ready the Tx's
1597
1598             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1599                     cohort1, modification1, true, false), getRef());
1600             expectMsgClass(duration, ReadyTransactionReply.class);
1601
1602             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1603                     cohort2, modification2, true, false), getRef());
1604             expectMsgClass(duration, ReadyTransactionReply.class);
1605
1606             // canCommit 1st Tx. We don't send the commit so it should timeout.
1607
1608             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1609             expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
1610
1611             // canCommit the 2nd Tx - it should complete after the 1st Tx times out.
1612
1613             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1614             expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
1615
1616             // Try to commit the 1st Tx - should fail as it's not the current Tx.
1617
1618             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
1619             expectMsgClass(duration, akka.actor.Status.Failure.class);
1620
1621             // Commit the 2nd Tx.
1622
1623             shard.tell(new CommitTransaction(transactionID2).toSerializable(), getRef());
1624             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1625
1626             NormalizedNode<?, ?> node = readStore(shard, listNodePath);
1627             assertNotNull(listNodePath + " not found", node);
1628
1629             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1630         }};
1631     }
1632
1633     @Test
1634     public void testTransactionCommitQueueCapacityExceeded() throws Throwable {
1635         dataStoreContextBuilder.shardTransactionCommitQueueCapacity(1);
1636
1637         new ShardTestKit(getSystem()) {{
1638             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1639                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1640                     "testTransactionCommitQueueCapacityExceeded");
1641
1642             waitUntilLeader(shard);
1643
1644             final FiniteDuration duration = duration("5 seconds");
1645
1646             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
1647
1648             String transactionID1 = "tx1";
1649             MutableCompositeModification modification1 = new MutableCompositeModification();
1650             ShardDataTreeCohort cohort1 = setupMockWriteTransaction("cohort1", dataStore,
1651                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification1);
1652
1653             String transactionID2 = "tx2";
1654             MutableCompositeModification modification2 = new MutableCompositeModification();
1655             ShardDataTreeCohort cohort2 = setupMockWriteTransaction("cohort2", dataStore,
1656                     TestModel.OUTER_LIST_PATH,
1657                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(),
1658                     modification2);
1659
1660             String transactionID3 = "tx3";
1661             MutableCompositeModification modification3 = new MutableCompositeModification();
1662             ShardDataTreeCohort cohort3 = setupMockWriteTransaction("cohort3", dataStore,
1663                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification3);
1664
1665             // Ready the Tx's
1666
1667             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1668                     cohort1, modification1, true, false), getRef());
1669             expectMsgClass(duration, ReadyTransactionReply.class);
1670
1671             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1672                     cohort2, modification2, true, false), getRef());
1673             expectMsgClass(duration, ReadyTransactionReply.class);
1674
1675             shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION,
1676                     cohort3, modification3, true, false), getRef());
1677             expectMsgClass(duration, ReadyTransactionReply.class);
1678
1679             // canCommit 1st Tx.
1680
1681             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1682             expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
1683
1684             // canCommit the 2nd Tx - it should get queued.
1685
1686             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1687
1688             // canCommit the 3rd Tx - should exceed queue capacity and fail.
1689
1690             shard.tell(new CanCommitTransaction(transactionID3).toSerializable(), getRef());
1691             expectMsgClass(duration, akka.actor.Status.Failure.class);
1692
1693             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1694         }};
1695     }
1696
1697     @Test
1698     public void testCanCommitBeforeReadyFailure() throws Throwable {
1699         new ShardTestKit(getSystem()) {{
1700             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1701                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1702                     "testCanCommitBeforeReadyFailure");
1703
1704             shard.tell(new CanCommitTransaction("tx").toSerializable(), getRef());
1705             expectMsgClass(duration("5 seconds"), akka.actor.Status.Failure.class);
1706
1707             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1708         }};
1709     }
1710
1711     @Test
1712     public void testAbortTransaction() throws Throwable {
1713         new ShardTestKit(getSystem()) {{
1714             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1715                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1716                     "testAbortTransaction");
1717
1718             waitUntilLeader(shard);
1719
1720             // Setup 2 simulated transactions with mock cohorts. The first one will be aborted.
1721
1722             String transactionID1 = "tx1";
1723             MutableCompositeModification modification1 = new MutableCompositeModification();
1724             ShardDataTreeCohort cohort1 = mock(ShardDataTreeCohort.class, "cohort1");
1725             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort1).canCommit();
1726             doReturn(Futures.immediateFuture(null)).when(cohort1).abort();
1727
1728             String transactionID2 = "tx2";
1729             MutableCompositeModification modification2 = new MutableCompositeModification();
1730             ShardDataTreeCohort cohort2 = mock(ShardDataTreeCohort.class, "cohort2");
1731             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort2).canCommit();
1732
1733             FiniteDuration duration = duration("5 seconds");
1734             final Timeout timeout = new Timeout(duration);
1735
1736             // Simulate the ForwardedReadyTransaction messages that would be sent
1737             // by the ShardTransaction.
1738
1739             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1740                     cohort1, modification1, true, false), getRef());
1741             expectMsgClass(duration, ReadyTransactionReply.class);
1742
1743             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1744                     cohort2, modification2, true, false), getRef());
1745             expectMsgClass(duration, ReadyTransactionReply.class);
1746
1747             // Send the CanCommitTransaction message for the first Tx.
1748
1749             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1750             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1751                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1752             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1753
1754             // Send the CanCommitTransaction message for the 2nd Tx. This should get queued and
1755             // processed after the first Tx completes.
1756
1757             Future<Object> canCommitFuture = Patterns.ask(shard,
1758                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
1759
1760             // Send the AbortTransaction message for the first Tx. This should trigger the 2nd
1761             // Tx to proceed.
1762
1763             shard.tell(new AbortTransaction(transactionID1).toSerializable(), getRef());
1764             expectMsgClass(duration, AbortTransactionReply.SERIALIZABLE_CLASS);
1765
1766             // Wait for the 2nd Tx to complete the canCommit phase.
1767
1768             Await.ready(canCommitFuture, duration);
1769
1770             InOrder inOrder = inOrder(cohort1, cohort2);
1771             inOrder.verify(cohort1).canCommit();
1772             inOrder.verify(cohort2).canCommit();
1773
1774             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1775         }};
1776     }
1777
1778     @Test
1779     public void testCreateSnapshot() throws Exception {
1780         testCreateSnapshot(true, "testCreateSnapshot");
1781     }
1782
1783     @Test
1784     public void testCreateSnapshotWithNonPersistentData() throws Exception {
1785         testCreateSnapshot(false, "testCreateSnapshotWithNonPersistentData");
1786     }
1787
1788     @SuppressWarnings("serial")
1789     public void testCreateSnapshot(final boolean persistent, final String shardActorName) throws Exception{
1790
1791         final AtomicReference<CountDownLatch> latch = new AtomicReference<>(new CountDownLatch(1));
1792
1793         final AtomicReference<Object> savedSnapshot = new AtomicReference<>();
1794         class TestPersistentDataProvider extends DelegatingPersistentDataProvider {
1795             TestPersistentDataProvider(DataPersistenceProvider delegate) {
1796                 super(delegate);
1797             }
1798
1799             @Override
1800             public void saveSnapshot(Object o) {
1801                 savedSnapshot.set(o);
1802                 super.saveSnapshot(o);
1803             }
1804         }
1805
1806         dataStoreContextBuilder.persistent(persistent);
1807
1808         new ShardTestKit(getSystem()) {{
1809             class TestShard extends Shard {
1810
1811                 protected TestShard(ShardIdentifier name, Map<String, String> peerAddresses,
1812                                     DatastoreContext datastoreContext, SchemaContext schemaContext) {
1813                     super(name, peerAddresses, datastoreContext, schemaContext);
1814                     setPersistence(new TestPersistentDataProvider(super.persistence()));
1815                 }
1816
1817                 @Override
1818                 public void handleCommand(Object message) {
1819                     super.handleCommand(message);
1820
1821                     if (message instanceof SaveSnapshotSuccess || message.equals("commit_snapshot")) {
1822                         latch.get().countDown();
1823                     }
1824                 }
1825
1826                 @Override
1827                 public RaftActorContext getRaftActorContext() {
1828                     return super.getRaftActorContext();
1829                 }
1830             }
1831
1832             Creator<Shard> creator = new Creator<Shard>() {
1833                 @Override
1834                 public Shard create() throws Exception {
1835                     return new TestShard(shardID, Collections.<String,String>emptyMap(),
1836                             newDatastoreContext(), SCHEMA_CONTEXT);
1837                 }
1838             };
1839
1840             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1841                     Props.create(new DelegatingShardCreator(creator)), shardActorName);
1842
1843             waitUntilLeader(shard);
1844
1845             writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1846
1847             NormalizedNode<?,?> expectedRoot = readStore(shard, YangInstanceIdentifier.builder().build());
1848
1849             // Trigger creation of a snapshot by ensuring
1850             RaftActorContext raftActorContext = ((TestShard) shard.underlyingActor()).getRaftActorContext();
1851             raftActorContext.getSnapshotManager().capture(mock(ReplicatedLogEntry.class), -1);
1852
1853             assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
1854
1855             assertTrue("Invalid saved snapshot " + savedSnapshot.get(),
1856                     savedSnapshot.get() instanceof Snapshot);
1857
1858             verifySnapshot((Snapshot)savedSnapshot.get(), expectedRoot);
1859
1860             latch.set(new CountDownLatch(1));
1861             savedSnapshot.set(null);
1862
1863             raftActorContext.getSnapshotManager().capture(mock(ReplicatedLogEntry.class), -1);
1864
1865             assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
1866
1867             assertTrue("Invalid saved snapshot " + savedSnapshot.get(),
1868                     savedSnapshot.get() instanceof Snapshot);
1869
1870             verifySnapshot((Snapshot)savedSnapshot.get(), expectedRoot);
1871
1872             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1873         }
1874
1875         private void verifySnapshot(Snapshot snapshot, NormalizedNode<?,?> expectedRoot) {
1876
1877             NormalizedNode<?, ?> actual = SerializationUtils.deserializeNormalizedNode(snapshot.getState());
1878             assertEquals("Root node", expectedRoot, actual);
1879
1880         }};
1881     }
1882
1883     /**
1884      * This test simply verifies that the applySnapShot logic will work
1885      * @throws ReadFailedException
1886      * @throws DataValidationFailedException
1887      */
1888     @Test
1889     public void testInMemoryDataTreeRestore() throws ReadFailedException, DataValidationFailedException {
1890         DataTree store = InMemoryDataTreeFactory.getInstance().create();
1891         store.setSchemaContext(SCHEMA_CONTEXT);
1892
1893         DataTreeModification putTransaction = store.takeSnapshot().newModification();
1894         putTransaction.write(TestModel.TEST_PATH,
1895             ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1896         commitTransaction(store, putTransaction);
1897
1898
1899         NormalizedNode<?, ?> expected = readStore(store, YangInstanceIdentifier.builder().build());
1900
1901         DataTreeModification writeTransaction = store.takeSnapshot().newModification();
1902
1903         writeTransaction.delete(YangInstanceIdentifier.builder().build());
1904         writeTransaction.write(YangInstanceIdentifier.builder().build(), expected);
1905
1906         commitTransaction(store, writeTransaction);
1907
1908         NormalizedNode<?, ?> actual = readStore(store, YangInstanceIdentifier.builder().build());
1909
1910         assertEquals(expected, actual);
1911     }
1912
1913     @Test
1914     public void testRecoveryApplicable(){
1915
1916         final DatastoreContext persistentContext = DatastoreContext.newBuilder().
1917                 shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(true).build();
1918
1919         final Props persistentProps = Shard.props(shardID, Collections.<String, String>emptyMap(),
1920                 persistentContext, SCHEMA_CONTEXT);
1921
1922         final DatastoreContext nonPersistentContext = DatastoreContext.newBuilder().
1923                 shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(false).build();
1924
1925         final Props nonPersistentProps = Shard.props(shardID, Collections.<String, String>emptyMap(),
1926                 nonPersistentContext, SCHEMA_CONTEXT);
1927
1928         new ShardTestKit(getSystem()) {{
1929             TestActorRef<Shard> shard1 = TestActorRef.create(getSystem(),
1930                     persistentProps, "testPersistence1");
1931
1932             assertTrue("Recovery Applicable", shard1.underlyingActor().persistence().isRecoveryApplicable());
1933
1934             shard1.tell(PoisonPill.getInstance(), ActorRef.noSender());
1935
1936             TestActorRef<Shard> shard2 = TestActorRef.create(getSystem(),
1937                     nonPersistentProps, "testPersistence2");
1938
1939             assertFalse("Recovery Not Applicable", shard2.underlyingActor().persistence().isRecoveryApplicable());
1940
1941             shard2.tell(PoisonPill.getInstance(), ActorRef.noSender());
1942
1943         }};
1944
1945     }
1946
1947     @Test
1948     public void testOnDatastoreContext() {
1949         new ShardTestKit(getSystem()) {{
1950             dataStoreContextBuilder.persistent(true);
1951
1952             TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(), "testOnDatastoreContext");
1953
1954             assertEquals("isRecoveryApplicable", true,
1955                     shard.underlyingActor().persistence().isRecoveryApplicable());
1956
1957             waitUntilLeader(shard);
1958
1959             shard.tell(dataStoreContextBuilder.persistent(false).build(), ActorRef.noSender());
1960
1961             assertEquals("isRecoveryApplicable", false,
1962                     shard.underlyingActor().persistence().isRecoveryApplicable());
1963
1964             shard.tell(dataStoreContextBuilder.persistent(true).build(), ActorRef.noSender());
1965
1966             assertEquals("isRecoveryApplicable", true,
1967                     shard.underlyingActor().persistence().isRecoveryApplicable());
1968
1969             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1970         }};
1971     }
1972
1973     @Test
1974     public void testRegisterRoleChangeListener() throws Exception {
1975         new ShardTestKit(getSystem()) {
1976             {
1977                 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1978                         newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1979                         "testRegisterRoleChangeListener");
1980
1981                 waitUntilLeader(shard);
1982
1983                 TestActorRef<MessageCollectorActor> listener =
1984                         TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class));
1985
1986                 shard.tell(new RegisterRoleChangeListener(), listener);
1987
1988                 // TODO: MessageCollectorActor exists as a test util in both the akka-raft and distributed-datastore
1989                 // projects. Need to move it to commons as a regular utility and then we can get rid of this arbitrary
1990                 // sleep.
1991                 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
1992
1993                 List<Object> allMatching = MessageCollectorActor.getAllMatching(listener, RegisterRoleChangeListenerReply.class);
1994
1995                 assertEquals(1, allMatching.size());
1996             }
1997         };
1998     }
1999
2000     @Test
2001     public void testFollowerInitialSyncStatus() throws Exception {
2002         final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
2003                 newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
2004                 "testFollowerInitialSyncStatus");
2005
2006         shard.underlyingActor().onReceiveCommand(new FollowerInitialSyncUpStatus(false, "member-1-shard-inventory-operational"));
2007
2008         assertEquals(false, shard.underlyingActor().getShardMBean().getFollowerInitialSyncStatus());
2009
2010         shard.underlyingActor().onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-inventory-operational"));
2011
2012         assertEquals(true, shard.underlyingActor().getShardMBean().getFollowerInitialSyncStatus());
2013
2014         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
2015     }
2016
2017     private static void commitTransaction(DataTree store, final DataTreeModification modification) throws DataValidationFailedException {
2018         modification.ready();
2019         store.validate(modification);
2020         store.commit(store.prepare(modification));
2021     }
2022 }