Shard extends RaftActor { @Override protected void applyState(final ActorRef clientActor, final Identifier identifier, final Object data) { if (data instanceof Payload) { if (data instanceof DisableTrackingPayload) { disableTracking((DisableTrackingPayload) data); return; } try { store.applyReplicatedPayload(identifier, (Payload)data); } catch (DataValidationFailedException | IOException e) { LOG.error("{}: Error applying replica {}", persistenceId(), identifier, e); } } else { LOG.error("{}: Unknown state for {} received {}", persistenceId(), identifier, data); } } 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 testApplyState() throws Exception { final TestActorRef<Shard> shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), "testApplyState"); ShardTestKit.waitUntilLeader(shard); final DataTree store = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_OPERATIONAL, SCHEMA_CONTEXT); final DataTreeModification writeMod = store.takeSnapshot().newModification(); final ContainerNode node = ImmutableNodes.containerNode(TestModel.TEST_QNAME); writeMod.write(TestModel.TEST_PATH, node); writeMod.ready(); final TransactionIdentifier tx = nextTransactionId(); shard.underlyingActor().applyState(null, null, payloadForModification(store, writeMod, tx)); final Stopwatch sw = Stopwatch.createStarted(); while (sw.elapsed(TimeUnit.SECONDS) <= 5) { Uninterruptibles.sleepUninterruptibly(75, TimeUnit.MILLISECONDS); final NormalizedNode<?,?> actual = readStore(shard, TestModel.TEST_PATH); if (actual != null) { assertEquals("Applied state", node, actual); return; } } fail("State was not applied"); }