Shard extends RaftActor { public int getPendingTxCommitQueueSize() { return store.getQueueSize(); } 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 testAbortAfterReady() { dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1); final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testAbortAfterReady"); ShardTestKit.waitUntilLeader(shard); final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID1 = nextTransactionId(); shard.tell(newBatchedModifications(transactionID1, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), true, false, 1), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new AbortTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, AbortTransactionReply.class); assertEquals("getPendingTxCommitQueueSize", 0, shard.underlyingActor().getPendingTxCommitQueueSize()); shard.tell(new CanCommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); final Throwable failure = testKit.expectMsgClass(duration, akka.actor.Status.Failure.class).cause(); assertTrue("Failure type", failure instanceof IllegalStateException); final TransactionIdentifier transactionID2 = nextTransactionId(); shard.tell(newBatchedModifications(transactionID2, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), true, false, 1), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID2, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CanCommitTransactionReply.class); }
@Test public void testAbortQueuedTransaction() { final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testAbortAfterReady"); ShardTestKit.waitUntilLeader(shard); final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID1 = nextTransactionId(); shard.tell(newBatchedModifications(transactionID1, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), true, false, 1), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); final TransactionIdentifier transactionID2 = nextTransactionId(); shard.tell(newBatchedModifications(transactionID2, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), true, false, 1), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); final TransactionIdentifier transactionID3 = nextTransactionId(); shard.tell(newBatchedModifications(transactionID3, TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), true, false, 1), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new AbortTransaction(transactionID2, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, AbortTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CanCommitTransactionReply.class); shard.tell(new CommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CommitTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID3, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CanCommitTransactionReply.class); shard.tell(new CommitTransaction(transactionID3, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CommitTransactionReply.class); assertEquals("getPendingTxCommitQueueSize", 0, shard.underlyingActor().getPendingTxCommitQueueSize()); }