@Test @SuppressWarnings("checkstyle:IllegalCatch") public void testConcurrentThreePhaseCommits() throws Exception { final AtomicReference<Throwable> caughtEx = new AtomicReference<>(); final CountDownLatch commitLatch = new CountDownLatch(2); final long timeoutSec = 5; final Duration duration = Duration.ofSeconds(timeoutSec); final Timeout timeout = Timeout.create(duration); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testConcurrentThreePhaseCommits"); class OnFutureComplete extends OnComplete<Object> { private final Class<?> expRespType; OnFutureComplete(final Class<?> expRespType) { this.expRespType = expRespType; } @Override public void onComplete(final Throwable error, final Object resp) { if (error != null) { caughtEx.set(new AssertionError(getClass().getSimpleName() + " failure", error)); } else { try { assertEquals("Commit response type", expRespType, resp.getClass()); onSuccess(resp); } catch (final Exception e) { caughtEx.set(e); } } } void onSuccess(final Object resp) { } } class OnCommitFutureComplete extends OnFutureComplete { OnCommitFutureComplete() { super(CommitTransactionReply.class); } @Override public void onComplete(final Throwable error, final Object resp) { super.onComplete(error, resp); commitLatch.countDown(); } } class OnCanCommitFutureComplete extends OnFutureComplete { private final TransactionIdentifier transactionID; OnCanCommitFutureComplete(final TransactionIdentifier transactionID) { super(CanCommitTransactionReply.class); this.transactionID = transactionID; } @Override void onSuccess(final Object resp) { final CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(resp); assertTrue("Can commit", canCommitReply.getCanCommit()); final Future<Object> commitFuture = Patterns.ask(shard, new CommitTransaction(transactionID, CURRENT_VERSION).toSerializable(), timeout); commitFuture.onComplete(new OnCommitFutureComplete(), getSystem().dispatcher()); } } final ShardTestKit testKit = new ShardTestKit(getSystem()); ShardTestKit.waitUntilLeader(shard); final TransactionIdentifier transactionID1 = nextTransactionId(); final TransactionIdentifier transactionID2 = nextTransactionId(); final TransactionIdentifier transactionID3 = nextTransactionId(); final Map<TransactionIdentifier, CapturingShardDataTreeCohort> cohortMap = setupCohortDecorator( shard.underlyingActor(), transactionID1, transactionID2, transactionID3); final CapturingShardDataTreeCohort cohort1 = cohortMap.get(transactionID1); final CapturingShardDataTreeCohort cohort2 = cohortMap.get(transactionID2); final CapturingShardDataTreeCohort cohort3 = cohortMap.get(transactionID3); shard.tell(prepareBatchedModifications(transactionID1, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), false), testKit.getRef()); final ReadyTransactionReply readyReply = ReadyTransactionReply .fromSerializable(testKit.expectMsgClass(duration, ReadyTransactionReply.class)); assertEquals("Cohort path", shard.path().toString(), readyReply.getCohortPath()); shard.tell(new CanCommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); final CanCommitTransactionReply canCommitReply = CanCommitTransactionReply .fromSerializable(testKit.expectMsgClass(duration, CanCommitTransactionReply.class)); assertTrue("Can commit", canCommitReply.getCanCommit()); shard.tell(prepareBatchedModifications(transactionID2, TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); shard.tell( prepareBatchedModifications(transactionID3, YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(), ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1), false), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); final Future<Object> canCommitFuture1 = Patterns.ask(shard, new CanCommitTransaction(transactionID2, CURRENT_VERSION).toSerializable(), timeout); final Future<Object> canCommitFuture2 = Patterns.ask(shard, new CanCommitTransaction(transactionID3, CURRENT_VERSION).toSerializable(), timeout); shard.tell(new CommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CommitTransactionReply.class); canCommitFuture1.onComplete(new OnCanCommitFutureComplete(transactionID2), getSystem().dispatcher()); canCommitFuture2.onComplete(new OnCanCommitFutureComplete(transactionID3), getSystem().dispatcher()); final boolean done = commitLatch.await(timeoutSec, TimeUnit.SECONDS); final Throwable t = caughtEx.get(); if (t != null) { Throwables.propagateIfPossible(t, Exception.class); throw new RuntimeException(t); } assertTrue("Commits complete", done); verifyOuterListEntry(shard, 1); verifyLastApplied(shard, 3); }
public static Builder builder() { return new Builder(); }
Shard extends RaftActor { public static Builder builder() { return new Builder(); } }
Shard extends RaftActor { public static Builder builder() { return new Builder(); } protected  Shard(final AbstractBuilder<?, ?> builder); }
Shard extends RaftActor { public static Builder builder() { return new Builder(); } 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(); }
Shard extends RaftActor { public static Builder builder() { return new Builder(); } 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; }