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; }
@Test public void testDataTreeCandidateRecovery() throws Exception { final DataTree source = setupInMemorySnapshotStore(); final DataTreeModification writeMod = source.takeSnapshot().newModification(); writeMod.write(TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build()); writeMod.ready(); InMemoryJournal.addEntry(shardID.toString(), 0, DUMMY_DATA); InMemoryJournal.addEntry(shardID.toString(), 1, new SimpleReplicatedLogEntry(0, 1, payloadForModification(source, writeMod, nextTransactionId()))); final int nListEntries = 16; final Set<Integer> listEntryKeys = new HashSet<>(); for (int i = 1; i <= nListEntries; i++) { listEntryKeys.add(Integer.valueOf(i)); final YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build(); final DataTreeModification mod = source.takeSnapshot().newModification(); mod.merge(path, ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i)); mod.ready(); InMemoryJournal.addEntry(shardID.toString(), i + 1, new SimpleReplicatedLogEntry(i, 1, payloadForModification(source, mod, nextTransactionId()))); } InMemoryJournal.addEntry(shardID.toString(), nListEntries + 2, new ApplyJournalEntries(nListEntries)); testRecovery(listEntryKeys); }
@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); }
@Test public void testBatchedModificationsWithNoCommitOnReady() { final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testBatchedModificationsWithNoCommitOnReady"); ShardTestKit.waitUntilLeader(shard); final TransactionIdentifier transactionID = nextTransactionId(); final Duration duration = Duration.ofSeconds(5); shard.tell(newBatchedModifications(transactionID, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), false, false, 1), testKit.getRef()); testKit.expectMsgClass(duration, BatchedModificationsReply.class); shard.tell(newBatchedModifications(transactionID, TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false, 2), testKit.getRef()); testKit.expectMsgClass(duration, BatchedModificationsReply.class); shard.tell(newBatchedModifications(transactionID, 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), true, false, 3), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID, CURRENT_VERSION).toSerializable(), testKit.getRef()); final CanCommitTransactionReply canCommitReply = CanCommitTransactionReply .fromSerializable(testKit.expectMsgClass(duration, CanCommitTransactionReply.class)); assertTrue("Can commit", canCommitReply.getCanCommit()); shard.tell(new CommitTransaction(transactionID, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CommitTransactionReply.class); verifyOuterListEntry(shard, 1); }
@Test public void testBatchedModificationsWithCommitOnReady() { final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testBatchedModificationsWithCommitOnReady"); ShardTestKit.waitUntilLeader(shard); final TransactionIdentifier transactionID = nextTransactionId(); final Duration duration = Duration.ofSeconds(5); shard.tell(newBatchedModifications(transactionID, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), false, false, 1), testKit.getRef()); testKit.expectMsgClass(duration, BatchedModificationsReply.class); shard.tell(newBatchedModifications(transactionID, TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false, 2), testKit.getRef()); testKit.expectMsgClass(duration, BatchedModificationsReply.class); shard.tell(newBatchedModifications(transactionID, 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), true, true, 3), testKit.getRef()); testKit.expectMsgClass(duration, CommitTransactionReply.class); verifyOuterListEntry(shard, 1); }
@Test public void testTransactionCommitTimeout() throws Exception { dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1); final ShardTestKit testKit = new ShardTestKit(getSystem()); final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testTransactionCommitTimeout"); ShardTestKit.waitUntilLeader(shard); final Duration duration = Duration.ofSeconds(5); writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); writeToStore(shard, TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build()); final TransactionIdentifier transactionID1 = nextTransactionId(); shard.tell( prepareBatchedModifications(transactionID1, 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 TransactionIdentifier transactionID2 = nextTransactionId(); final YangInstanceIdentifier listNodePath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(); shard.tell( prepareBatchedModifications(transactionID2, listNodePath, ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2), false), testKit.getRef()); testKit.expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CanCommitTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID2, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CanCommitTransactionReply.class); shard.tell(new CommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, akka.actor.Status.Failure.class); shard.tell(new CommitTransaction(transactionID2, CURRENT_VERSION).toSerializable(), testKit.getRef()); testKit.expectMsgClass(duration, CommitTransactionReply.class); final NormalizedNode<?, ?> node = readStore(shard, listNodePath); assertNotNull(listNodePath + " not found", node); }
@Test public void testRecoveryApplicable() { final DatastoreContext persistentContext = DatastoreContext.newBuilder() .shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(true).build(); final Props persistentProps = Shard.builder().id(shardID).datastoreContext(persistentContext) .schemaContextProvider(() -> SCHEMA_CONTEXT).props(); final DatastoreContext nonPersistentContext = DatastoreContext.newBuilder() .shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(false).build(); final Props nonPersistentProps = Shard.builder().id(shardID).datastoreContext(nonPersistentContext) .schemaContextProvider(() -> SCHEMA_CONTEXT).props(); final TestActorRef<Shard> shard1 = actorFactory.createTestActor(persistentProps, "testPersistence1"); assertTrue("Recovery Applicable", shard1.underlyingActor().persistence().isRecoveryApplicable()); final TestActorRef<Shard> shard2 = actorFactory.createTestActor(nonPersistentProps, "testPersistence2"); assertFalse("Recovery Not Applicable", shard2.underlyingActor().persistence().isRecoveryApplicable()); }
@Test public void testClusteredDataTreeChangeListenerRegistration() throws Exception { final ShardTestKit testKit = new ShardTestKit(getSystem()); final String testName = "testClusteredDataTreeChangeListenerRegistration"; final ShardIdentifier followerShardID = ShardIdentifier.create("inventory", MemberName.forName(actorFactory.generateActorId(testName + "-follower")), "config"); final ShardIdentifier leaderShardID = ShardIdentifier.create("inventory", MemberName.forName(actorFactory.generateActorId(testName + "-leader")), "config"); final TestActorRef<Shard> followerShard = actorFactory .createTestActor(Shard.builder().id(followerShardID) .datastoreContext(dataStoreContextBuilder.shardElectionTimeoutFactor(1000).build()) .peerAddresses(Collections.singletonMap(leaderShardID.toString(), "akka: .schemaContextProvider(() -> SCHEMA_CONTEXT).props() .withDispatcher(Dispatchers.DefaultDispatcherId()), followerShardID.toString()); final TestActorRef<Shard> leaderShard = actorFactory .createTestActor(Shard.builder().id(leaderShardID).datastoreContext(newDatastoreContext()) .peerAddresses(Collections.singletonMap(followerShardID.toString(), "akka: .schemaContextProvider(() -> SCHEMA_CONTEXT).props() .withDispatcher(Dispatchers.DefaultDispatcherId()), leaderShardID.toString()); leaderShard.tell(TimeoutNow.INSTANCE, ActorRef.noSender()); final String leaderPath = ShardTestKit.waitUntilLeader(followerShard); assertEquals("Shard leader path", leaderShard.path().toString(), leaderPath); final YangInstanceIdentifier path = TestModel.TEST_PATH; final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1); final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener, path), actorFactory.generateActorId(testName + "-DataTreeChangeListener")); followerShard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), testKit.getRef()); final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); writeToStore(followerShard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); listener.waitForChangeEvents(); }