ShardManager extends AbstractUntypedPersistentActorWithMetering { @SuppressWarnings("checkstyle:IllegalCatch") private void onCreateShard(final CreateShard createShard) { LOG.debug("{}: onCreateShard: {}", persistenceId(), createShard); Object reply; try { String shardName = createShard.getModuleShardConfig().getShardName(); if (localShards.containsKey(shardName)) { LOG.debug("{}: Shard {} already exists", persistenceId(), shardName); reply = new Status.Success(String.format("Shard with name %s already exists", shardName)); } else { doCreateShard(createShard); reply = new Status.Success(null); } } catch (Exception e) { LOG.error("{}: onCreateShard failed", persistenceId(), e); reply = new Status.Failure(e); } if (getSender() != null && !getContext().system().deadLetters().equals(getSender())) { getSender().tell(reply, getSelf()); } } ShardManager(final AbstractShardManagerCreator<?> builder); @Override void preStart(); @Override void postStop(); @Override void handleCommand(final Object message); @Override SupervisorStrategy supervisorStrategy(); @Override String persistenceId();  }
@Test public void testOnCreateShard() { LOG.info("testOnCreateShard starting"); final TestKit kit = new TestKit(getSystem()); datastoreContextBuilder.shardInitializationTimeout(1, TimeUnit.MINUTES).persistent(true); ActorRef shardManager = actorFactory .createActor(newShardMgrProps(new ConfigurationImpl(new EmptyModuleShardConfigProvider())) .withDispatcher(Dispatchers.DefaultDispatcherId())); EffectiveModelContext schemaContext = TEST_SCHEMA_CONTEXT; shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender()); DatastoreContext datastoreContext = DatastoreContext.newBuilder().shardElectionTimeoutFactor(100) .persistent(false).build(); Shard.Builder shardBuilder = Shard.builder(); ModuleShardConfiguration config = new ModuleShardConfiguration(URI.create("foo-ns"), "foo-module", "foo", null, members("member-1", "member-5", "member-6")); shardManager.tell(new CreateShard(config, shardBuilder, datastoreContext), kit.getRef()); kit.expectMsgClass(Duration.ofSeconds(5), Success.class); shardManager.tell(new FindLocalShard("foo", true), kit.getRef()); kit.expectMsgClass(Duration.ofSeconds(5), LocalShardFound.class); assertFalse("isRecoveryApplicable", shardBuilder.getDatastoreContext().isPersistent()); assertTrue("Epxected ShardPeerAddressResolver", shardBuilder.getDatastoreContext().getShardRaftConfig() .getPeerAddressResolver() instanceof ShardPeerAddressResolver); assertEquals("peerMembers", Sets.newHashSet( ShardIdentifier.create("foo", MemberName.forName("member-5"), shardMrgIDSuffix).toString(), ShardIdentifier.create("foo", MemberName.forName("member-6"), shardMrgIDSuffix).toString()), shardBuilder.getPeerAddresses().keySet()); assertEquals("ShardIdentifier", ShardIdentifier.create("foo", MEMBER_1, shardMrgIDSuffix), shardBuilder.getId()); assertSame("schemaContext", schemaContext, shardBuilder.getSchemaContext()); shardManager.tell(new CreateShard(config, shardBuilder, null), kit.getRef()); Success success = kit.expectMsgClass(Duration.ofSeconds(5), Success.class); assertNotNull("Success status is null", success.status()); LOG.info("testOnCreateShard ending"); }