Shard extends RaftActor { @SuppressWarnings("checkstyle:IllegalCatch") private void createTransaction(final CreateTransaction createTransaction) { askProtocolEncountered(createTransaction.getTransactionId()); try { if (TransactionType.fromInt(createTransaction.getTransactionType()) != TransactionType.READ_ONLY && failIfIsolatedLeader(getSender())) { return; } ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(), createTransaction.getTransactionId()); getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), createTransaction.getTransactionId(), createTransaction.getVersion()).toSerializable(), getSelf()); } catch (Exception e) { getSender().tell(new Failure(e), getSelf()); } } 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 testCreateTransaction() { final ShardTestKit testKit = new ShardTestKit(getSystem()); final ActorRef shard = actorFactory.createActor(newShardProps(), "testCreateTransaction"); ShardTestKit.waitUntilLeader(shard); shard.tell(new UpdateSchemaContext(TestModel.createTestContext()), testKit.getRef()); shard.tell(new CreateTransaction(nextTransactionId(), TransactionType.READ_ONLY.ordinal(), DataStoreVersions.CURRENT_VERSION).toSerializable(), testKit.getRef()); final CreateTransactionReply reply = testKit.expectMsgClass(Duration.ofSeconds(3), CreateTransactionReply.class); final String path = reply.getTransactionPath().toString(); assertTrue("Unexpected transaction path " + path, path.startsWith(String.format( "akka: shardID.getShardName(), shardID.getMemberName().getName()))); }