+ @Test
+ public void testAddShardReplicaWithPreExistingLocalReplicaLeader() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
+ ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
+
+ shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+ shardManager.tell(new ActorInitialized(), mockShardActor);
+ shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION), getRef());
+ shardManager.tell((new RoleChangeNotification(memberId, RaftState.Candidate.name(),
+ RaftState.Leader.name())), mockShardActor);
+
+ shardManager.tell(new AddShardReplica(Shard.DEFAULT_NAME), getRef());
+ Failure resp = expectMsgClass(duration("5 seconds"), Failure.class);
+ assertEquals("Failure cause", AlreadyExistsException.class, resp.cause().getClass());
+
+ shardManager.tell(new FindLocalShard(Shard.DEFAULT_NAME, false), getRef());
+ expectMsgClass(duration("5 seconds"), LocalShardFound.class);
+ }};
+ }
+
+ @Test
+ public void testAddShardReplicaWithAddServerReplyFailure() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ JavaTestKit mockShardLeaderKit = new JavaTestKit(getSystem());
+
+ MockConfiguration mockConfig =
+ new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("astronauts", Arrays.asList("member-2")).build());
+
+ ActorRef mockNewReplicaShardActor = newMockShardActor(getSystem(), "astronauts", "member-1");
+ final TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(
+ newTestShardMgrBuilder(mockConfig).shardActor(mockNewReplicaShardActor).props(), shardMgrID);
+ shardManager.underlyingActor().setMessageInterceptor(newFindPrimaryInterceptor(mockShardLeaderKit.getRef()));
+
+ shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+
+ JavaTestKit terminateWatcher = new JavaTestKit(getSystem());
+ terminateWatcher.watch(mockNewReplicaShardActor);
+
+ shardManager.tell(new AddShardReplica("astronauts"), getRef());
+
+ AddServer addServerMsg = mockShardLeaderKit.expectMsgClass(AddServer.class);
+ assertEquals("AddServer serverId", "member-1-shard-astronauts-" + shardMrgIDSuffix,
+ addServerMsg.getNewServerId());
+ mockShardLeaderKit.reply(new AddServerReply(ServerChangeStatus.TIMEOUT, null));
+
+ Failure failure = expectMsgClass(duration("5 seconds"), Failure.class);
+ assertEquals("Failure cause", TimeoutException.class, failure.cause().getClass());
+
+ shardManager.tell(new FindLocalShard("astronauts", false), getRef());
+ expectMsgClass(duration("5 seconds"), LocalShardNotFound.class);
+
+ terminateWatcher.expectTerminated(mockNewReplicaShardActor);
+
+ shardManager.tell(new AddShardReplica("astronauts"), getRef());
+ mockShardLeaderKit.expectMsgClass(AddServer.class);
+ mockShardLeaderKit.reply(new AddServerReply(ServerChangeStatus.NO_LEADER, null));
+ failure = expectMsgClass(duration("5 seconds"), Failure.class);
+ assertEquals("Failure cause", NoShardLeaderException.class, failure.cause().getClass());
+ }};
+ }
+
+ @Test
+ public void testAddShardReplicaWithAlreadyInProgress() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ JavaTestKit mockShardLeaderKit = new JavaTestKit(getSystem());
+ JavaTestKit secondRequestKit = new JavaTestKit(getSystem());
+
+ MockConfiguration mockConfig =
+ new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("astronauts", Arrays.asList("member-2")).build());
+
+ final TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(
+ newTestShardMgrBuilder(mockConfig).shardActor(mockShardActor).props(), shardMgrID);
+ shardManager.underlyingActor().setMessageInterceptor(newFindPrimaryInterceptor(mockShardLeaderKit.getRef()));
+
+ shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+
+ shardManager.tell(new AddShardReplica("astronauts"), getRef());
+
+ mockShardLeaderKit.expectMsgClass(AddServer.class);
+
+ shardManager.tell(new AddShardReplica("astronauts"), secondRequestKit.getRef());
+
+ secondRequestKit.expectMsgClass(duration("5 seconds"), Failure.class);
+ }};
+ }
+
+ @Test
+ public void testAddShardReplicaWithFindPrimaryTimeout() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ MockConfiguration mockConfig = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("astronauts", Arrays.asList("member-2")).build());
+
+ final ActorRef newReplicaShardManager = actorFactory.createActor(newTestShardMgrBuilder(mockConfig).
+ shardActor(mockShardActor).props(), shardMgrID);
+
+ newReplicaShardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+ MockClusterWrapper.sendMemberUp(newReplicaShardManager, "member-2", getRef().path().toString());
+
+ newReplicaShardManager.tell(new AddShardReplica("astronauts"), getRef());
+ Status.Failure resp = expectMsgClass(duration("5 seconds"), Status.Failure.class);
+ assertEquals("Failure obtained", true,
+ (resp.cause() instanceof RuntimeException));
+ }};
+ }
+
+ @Test
+ public void testRemoveShardReplicaForNonExistentShard() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ ActorRef shardManager = actorFactory.createActor(newShardMgrProps(
+ new ConfigurationImpl(new EmptyModuleShardConfigProvider())));
+
+ shardManager.tell(new RemoveShardReplica("model-inventory"), getRef());
+ Status.Failure resp = expectMsgClass(duration("2 seconds"), Status.Failure.class);
+ assertEquals("Failure obtained", true,
+ (resp.cause() instanceof IllegalArgumentException));
+ }};
+
+ }
+
+ @Test
+ public void testServerRemovedShardActorNotRunning() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ MockConfiguration mockConfig =
+ new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("default", Arrays.asList("member-1", "member-2")).
+ put("astronauts", Arrays.asList("member-2")).
+ put("people", Arrays.asList("member-1", "member-2")).build());
+
+ TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(newShardMgrProps(mockConfig));
+
+ shardManager.underlyingActor().waitForRecoveryComplete();
+
+ shardManager.tell(new FindLocalShard("people", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+
+ shardManager.tell(new FindLocalShard("default", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+
+ // Removed the default shard replica from member-1
+ ShardIdentifier.Builder builder = new ShardIdentifier.Builder();
+ final ShardIdentifier shardId = builder.shardName("default").memberName("member-1").type("config1").build();
+ shardManager.tell(new ServerRemoved(shardId.toString()), getRef());
+
+ shardManager.underlyingActor().verifySnapshotPersisted(Sets.newHashSet("people"));
+ }};
+ }
+
+ @Test
+ public void testServerRemovedShardActorRunning() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ MockConfiguration mockConfig =
+ new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("default", Arrays.asList("member-1", "member-2")).
+ put("astronauts", Arrays.asList("member-2")).
+ put("people", Arrays.asList("member-1", "member-2")).build());
+
+ TestActorRef<MessageCollectorActor> shard = actorFactory.createTestActor(MessageCollectorActor.props());
+
+ TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(
+ newTestShardMgrBuilder(mockConfig).addShardActor("default", shard).props());
+
+ watch(shard);
+
+ shardManager.underlyingActor().waitForRecoveryComplete();
+
+ shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+
+ shardManager.tell(new FindLocalShard("people", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+
+ shardManager.tell(new FindLocalShard("default", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+
+ // Removed the default shard replica from member-1
+ ShardIdentifier.Builder builder = new ShardIdentifier.Builder();
+ final ShardIdentifier shardId = builder.shardName("default").memberName("member-1").type("config1").build();
+ shardManager.tell(new ServerRemoved(shardId.toString()), getRef());
+
+ shardManager.underlyingActor().verifySnapshotPersisted(Sets.newHashSet("people"));
+
+ expectMsgClass(duration("5 seconds"), Terminated.class);
+ }};
+ }
+
+
+ @Test
+ public void testShardPersistenceWithRestoredData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ MockConfiguration mockConfig =
+ new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("default", Arrays.asList("member-1", "member-2")).
+ put("astronauts", Arrays.asList("member-2")).
+ put("people", Arrays.asList("member-1", "member-2")).build());
+ String[] restoredShards = {"default", "astronauts"};
+ ShardManagerSnapshot snapshot = new ShardManagerSnapshot(Arrays.asList(restoredShards));
+ InMemorySnapshotStore.addSnapshot("shard-manager-" + shardMrgIDSuffix, snapshot);
+
+ //create shardManager to come up with restored data
+ TestActorRef<TestShardManager> newRestoredShardManager = actorFactory.createTestActor(
+ newShardMgrProps(mockConfig));
+
+ newRestoredShardManager.underlyingActor().waitForRecoveryComplete();
+
+ newRestoredShardManager.tell(new FindLocalShard("people", false), getRef());
+ LocalShardNotFound notFound = expectMsgClass(duration("5 seconds"), LocalShardNotFound.class);
+ assertEquals("for uninitialized shard", "people", notFound.getShardName());
+
+ //Verify a local shard is created for the restored shards,
+ //although we expect a NotInitializedException for the shards as the actor initialization
+ //message is not sent for them
+ newRestoredShardManager.tell(new FindLocalShard("default", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+
+ newRestoredShardManager.tell(new FindLocalShard("astronauts", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+ }};
+ }
+
+
+ private static class TestShardManager extends ShardManager {
+ private final CountDownLatch recoveryComplete = new CountDownLatch(1);
+ private final CountDownLatch snapshotPersist = new CountDownLatch(1);
+ private ShardManagerSnapshot snapshot;
+ private final Map<String, ActorRef> shardActors;
+ private final ActorRef shardActor;