Shard extends RaftActor { @VisibleForTesting public ShardDataTree getDataStore() { return store; } protected  Shard(final AbstractBuilder<?, ?> builder); @Override void postStop(); int getPendingTxCommitQueueSize(); int getCohortCacheSize(); @Override @VisibleForTesting RaftActorSnapshotCohort getRaftActorSnapshotCohort(); Collection<?> convertPendingTransactionsToMessages(); @Override String persistenceId(); DatastoreContext getDatastoreContext(); @VisibleForTesting ShardDataTree getDataStore(); static Builder builder(); static final String DEFAULT_NAME; }
@Test public void testReadyLocalTransactionWithImmediateCommit() { final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testReadyLocalTransactionWithImmediateCommit"); ShardTestKit.waitUntilLeader(shard); final ShardDataTree dataStore = shard.underlyingActor().getDataStore(); final DataTreeModification modification = dataStore.newModification(); final ContainerNode writeData = ImmutableNodes.containerNode(TestModel.TEST_QNAME); new WriteModification(TestModel.TEST_PATH, writeData).apply(modification); final MapNode mergeData = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME) .addChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42)) .build(); new MergeModification(TestModel.OUTER_LIST_PATH, mergeData).apply(modification); final TransactionIdentifier txId = nextTransactionId(); modification.ready(); final ReadyLocalTransaction readyMessage = new ReadyLocalTransaction(txId, modification, true, Optional.empty()); shard.tell(readyMessage, testKit.getRef()); testKit.expectMsgClass(CommitTransactionReply.class); final NormalizedNode<?, ?> actualNode = readStore(shard, TestModel.OUTER_LIST_PATH); assertEquals(TestModel.OUTER_LIST_QNAME.getLocalName(), mergeData, actualNode); }
@Test public void testReadyLocalTransactionWithThreePhaseCommit() { final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testReadyLocalTransactionWithThreePhaseCommit"); ShardTestKit.waitUntilLeader(shard); final ShardDataTree dataStore = shard.underlyingActor().getDataStore(); final DataTreeModification modification = dataStore.newModification(); final ContainerNode writeData = ImmutableNodes.containerNode(TestModel.TEST_QNAME); new WriteModification(TestModel.TEST_PATH, writeData).apply(modification); final MapNode mergeData = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME) .addChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42)) .build(); new MergeModification(TestModel.OUTER_LIST_PATH, mergeData).apply(modification); final TransactionIdentifier txId = nextTransactionId(); modification.ready(); final ReadyLocalTransaction readyMessage = new ReadyLocalTransaction(txId, modification, false, Optional.empty()); shard.tell(readyMessage, testKit.getRef()); testKit.expectMsgClass(ReadyTransactionReply.class); shard.tell(new CanCommitTransaction(txId, CURRENT_VERSION).toSerializable(), testKit.getRef()); final CanCommitTransactionReply canCommitReply = CanCommitTransactionReply .fromSerializable(testKit.expectMsgClass(CanCommitTransactionReply.class)); assertTrue("Can commit", canCommitReply.getCanCommit()); shard.tell(new CommitTransaction(txId, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(CommitTransactionReply.class); final NormalizedNode<?, ?> actualNode = readStore(shard, TestModel.OUTER_LIST_PATH); assertEquals(TestModel.OUTER_LIST_QNAME.getLocalName(), mergeData, actualNode); }
@Test public void testTransactionCommitWithSubsequentExpiredCohortEntry() { dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1); final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testTransactionCommitWithSubsequentExpiredCohortEntry"); ShardTestKit.waitUntilLeader(shard); final Duration duration = Duration.ofSeconds(5); final ShardDataTree dataStore = shard.underlyingActor().getDataStore(); final TransactionIdentifier transactionID1 = nextTransactionId(); shard.tell(prepareBatchedModifications(transactionID1, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), false), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CanCommitTransactionReply.class); final TransactionIdentifier transactionID2 = nextTransactionId(); shard.tell(prepareBatchedModifications(transactionID2, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), false), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); final TransactionIdentifier transactionID3 = nextTransactionId(); final DataTreeModification modification3 = dataStore.newModification(); new WriteModification(TestModel.TEST2_PATH, ImmutableNodes.containerNode(TestModel.TEST2_QNAME)) .apply(modification3); modification3.ready(); final ReadyLocalTransaction readyMessage = new ReadyLocalTransaction(transactionID3, modification3, true, Optional.empty()); shard.tell(readyMessage, testKit.getRef()); shard.tell(new CommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CommitTransactionReply.class); testKit.expectMsgClass(duration, CommitTransactionReply.class); final NormalizedNode<?, ?> node = readStore(shard, TestModel.TEST2_PATH); assertNotNull(TestModel.TEST2_PATH + " not found", node); }
@Test public void testRegisterRoleChangeListener() { final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testRegisterRoleChangeListener"); ShardTestKit.waitUntilLeader(shard); final ActorRef listener = getSystem().actorOf(MessageCollectorActor.props()); shard.tell(new RegisterRoleChangeListener(), listener); MessageCollectorActor.expectFirstMatching(listener, RegisterRoleChangeListenerReply.class); ShardLeaderStateChanged leaderStateChanged = MessageCollectorActor.expectFirstMatching(listener, ShardLeaderStateChanged.class); assertTrue("getLocalShardDataTree present", leaderStateChanged.getLocalShardDataTree().isPresent()); assertSame("getLocalShardDataTree", shard.underlyingActor().getDataStore().getDataTree(), leaderStateChanged.getLocalShardDataTree().get()); MessageCollectorActor.clearMessages(listener); shard.tell(new RequestVote(10000, "member2", 50, 50), testKit.getRef()); leaderStateChanged = MessageCollectorActor.expectFirstMatching(listener, ShardLeaderStateChanged.class); assertFalse("getLocalShardDataTree present", leaderStateChanged.getLocalShardDataTree().isPresent()); }