+ LOG.info("testOnReceiveFindPrimaryForRemoteShard ending");
+ }
+
+ @Test
+ public void testShardAvailabilityOnChangeOfMemberReachability() throws Exception {
+ LOG.info("testShardAvailabilityOnChangeOfMemberReachability starting");
+ String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
+
+ // Create an ActorSystem ShardManager actor for member-1.
+
+ final ActorSystem system1 = newActorSystem("Member1");
+ Cluster.get(system1).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ final ActorRef mockShardActor1 = newMockShardActor(system1, Shard.DEFAULT_NAME, "member-1");
+
+ final TestActorRef<TestShardManager> shardManager1 = TestActorRef.create(system1,
+ newTestShardMgrBuilder().shardActor(mockShardActor1).cluster(
+ new ClusterWrapperImpl(system1)).props(), shardManagerID);
+
+ // Create an ActorSystem ShardManager actor for member-2.
+
+ final ActorSystem system2 = newActorSystem("Member2");
+
+ Cluster.get(system2).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ final ActorRef mockShardActor2 = newMockShardActor(system2, Shard.DEFAULT_NAME, "member-2");
+
+ MockConfiguration mockConfig2 = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("default", Arrays.asList("member-1", "member-2")).build());
+
+ final TestActorRef<TestShardManager> shardManager2 = TestActorRef.create(system2,
+ newTestShardMgrBuilder(mockConfig2).shardActor(mockShardActor2).cluster(
+ new ClusterWrapperImpl(system2)).props(), shardManagerID);
+
+ new JavaTestKit(system1) {{
+
+ shardManager1.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+ shardManager2.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+ shardManager1.tell(new ActorInitialized(), mockShardActor1);
+ shardManager2.tell(new ActorInitialized(), mockShardActor2);
+
+ String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
+ String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId2,
+ Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION), mockShardActor1);
+ shardManager1.tell(new RoleChangeNotification(memberId1,
+ RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor1);
+ shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION),
+ mockShardActor2);
+ shardManager2.tell(new RoleChangeNotification(memberId2,
+ RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
+ shardManager1.underlyingActor().waitForMemberUp();
+
+ shardManager1.tell(new FindPrimary("default", true), getRef());
+
+ RemotePrimaryShardFound found = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
+ String path = found.getPrimaryPath();
+ assertTrue("Unexpected primary path " + path, path.contains("member-2-shard-default-config"));
+
+ shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
+ createUnreachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ shardManager1.underlyingActor().waitForUnreachableMember();
+
+ PeerDown peerDown = MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerDown.class);
+ assertEquals("getMemberName", "member-2", peerDown.getMemberName());
+ MessageCollectorActor.clearMessages(mockShardActor1);
+
+ shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
+ createMemberRemoved("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerDown.class);
+
+ shardManager1.tell(new FindPrimary("default", true), getRef());
+
+ expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
+
+ shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
+ createReachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ shardManager1.underlyingActor().waitForReachableMember();
+
+ PeerUp peerUp = MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerUp.class);
+ assertEquals("getMemberName", "member-2", peerUp.getMemberName());
+ MessageCollectorActor.clearMessages(mockShardActor1);
+
+ shardManager1.tell(new FindPrimary("default", true), getRef());
+
+ RemotePrimaryShardFound found1 = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
+ String path1 = found1.getPrimaryPath();
+ assertTrue("Unexpected primary path " + path1, path1.contains("member-2-shard-default-config"));
+
+ shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
+ createMemberUp("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerUp.class);
+
+ // Test FindPrimary wait succeeds after reachable member event.
+
+ shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
+ createUnreachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
+ shardManager1.underlyingActor().waitForUnreachableMember();
+
+ shardManager1.tell(new FindPrimary("default", true), getRef());
+
+ shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
+ createReachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ RemotePrimaryShardFound found2 = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
+ String path2 = found2.getPrimaryPath();
+ assertTrue("Unexpected primary path " + path2, path2.contains("member-2-shard-default-config"));
+ }};
+
+ LOG.info("testShardAvailabilityOnChangeOfMemberReachability ending");
+ }
+
+ @Test
+ public void testShardAvailabilityChangeOnMemberUnreachableAndLeadershipChange() throws Exception {
+ LOG.info("testShardAvailabilityChangeOnMemberUnreachableAndLeadershipChange starting");
+ String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
+
+ // Create an ActorSystem ShardManager actor for member-1.
+
+ final ActorSystem system1 = newActorSystem("Member1");
+ Cluster.get(system1).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ final ActorRef mockShardActor1 = newMockShardActor(system1, Shard.DEFAULT_NAME, "member-1");
+
+ final PrimaryShardInfoFutureCache primaryShardInfoCache = new PrimaryShardInfoFutureCache();
+ final TestActorRef<TestShardManager> shardManager1 = TestActorRef.create(system1,
+ newTestShardMgrBuilder().shardActor(mockShardActor1).cluster(
+ new ClusterWrapperImpl(system1)).primaryShardInfoCache(primaryShardInfoCache).props(),
+ shardManagerID);
+
+ // Create an ActorSystem ShardManager actor for member-2.
+
+ final ActorSystem system2 = newActorSystem("Member2");
+
+ Cluster.get(system2).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ final ActorRef mockShardActor2 = newMockShardActor(system2, Shard.DEFAULT_NAME, "member-2");
+
+ MockConfiguration mockConfig2 = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("default", Arrays.asList("member-1", "member-2")).build());
+
+ final TestActorRef<TestShardManager> shardManager2 = TestActorRef.create(system2,
+ newTestShardMgrBuilder(mockConfig2).shardActor(mockShardActor2).cluster(
+ new ClusterWrapperImpl(system2)).props(), shardManagerID);
+
+ new JavaTestKit(system1) {{
+ shardManager1.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+ shardManager2.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+ shardManager1.tell(new ActorInitialized(), mockShardActor1);
+ shardManager2.tell(new ActorInitialized(), mockShardActor2);
+
+ String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
+ String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId2,
+ Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION), mockShardActor1);
+ shardManager1.tell(new RoleChangeNotification(memberId1,
+ RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor1);
+ shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION),
+ mockShardActor2);
+ shardManager2.tell(new RoleChangeNotification(memberId2,
+ RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
+ shardManager1.underlyingActor().waitForMemberUp();
+
+ shardManager1.tell(new FindPrimary("default", true), getRef());
+
+ RemotePrimaryShardFound found = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
+ String path = found.getPrimaryPath();
+ assertTrue("Unexpected primary path " + path, path.contains("member-2-shard-default-config"));
+
+ primaryShardInfoCache.putSuccessful("default", new PrimaryShardInfo(system1.actorSelection(
+ mockShardActor1.path()), DataStoreVersions.CURRENT_VERSION, Optional.<DataTree>absent()));
+
+ shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
+ createUnreachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
+
+ shardManager1.underlyingActor().waitForUnreachableMember();
+
+ shardManager1.tell(new FindPrimary("default", true), getRef());
+
+ expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
+
+ assertNull("Expected primaryShardInfoCache entry removed", primaryShardInfoCache.getIfPresent("default"));
+
+ shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId1, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION), mockShardActor1);
+ shardManager1.tell(new RoleChangeNotification(memberId1,
+ RaftState.Follower.name(), RaftState.Leader.name()), mockShardActor1);
+
+ shardManager1.tell(new FindPrimary("default", true), getRef());
+
+ LocalPrimaryShardFound found1 = expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
+ String path1 = found1.getPrimaryPath();
+ assertTrue("Unexpected primary path " + path1, path1.contains("member-1-shard-default-config"));
+
+ }};
+
+ LOG.info("testShardAvailabilityChangeOnMemberUnreachableAndLeadershipChange ending");