ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica( final RemoveShardReplicaInput input) { final String shardName = input.getShardName(); if (Strings.isNullOrEmpty(shardName)) { return newFailedRpcResultFuture("A valid shard name must be specified"); } DataStoreType dataStoreType = input.getDataStoreType(); if (dataStoreType == null) { return newFailedRpcResultFuture("A valid DataStoreType must be specified"); } final String memberName = input.getMemberName(); if (Strings.isNullOrEmpty(memberName)) { return newFailedRpcResultFuture("A valid member name must be specified"); } LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType); final SettableFuture<RpcResult<RemoveShardReplicaOutput>> returnFuture = SettableFuture.create(); ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new RemoveShardReplica(shardName, MemberName.forName(memberName))); Futures.addCallback(future, new FutureCallback<Success>() { @Override public void onSuccess(final Success success) { LOG.info("Successfully removed replica for shard {}", shardName); returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build())); } @Override public void onFailure(final Throwable failure) { onMessageFailure(String.format("Failed to remove replica for shard %s", shardName), returnFuture, failure); } }, MoreExecutors.directExecutor()); return returnFuture; } ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
            final DistributedDataStoreInterface operDataStore,
            final BindingNormalizedNodeSerializer serializer); @Override ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input); @Override ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
            final RemoveShardReplicaInput input); @Override ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input); @Override ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input); @Override ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
            final AddPrefixShardReplicaInput input); @Override ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
            final RemovePrefixShardReplicaInput input); @Override ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
            final AddReplicasForAllShardsInput input); @Override ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
            final RemoveAllShardReplicasInput input); @Override ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
            final ChangeMemberVotingStatesForShardInput input); @Override ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
            final ChangeMemberVotingStatesForAllShardsInput input); @Override ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
            final FlipMemberVotingStatesForAllShardsInput input); @Override ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input); @Override ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
            final GetPrefixShardRoleInput input); @Override ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input); @Override ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
            final GetKnownClientsForAllShardsInput input);  }
@Test public void testRemoveShardReplica() throws Exception { String name = "testRemoveShardReplica"; String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; final MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder( DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)) .build(); final MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); final MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); leaderNode1.configDataStore().waitTillReady(); replicaNode3.configDataStore().waitTillReady(); verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3"); verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3"); verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2"); ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(), replicaNode3.operDataStore(), null); RpcResult<RemoveShardReplicaOutput> rpcResult = service3.removeShardReplica(new RemoveShardReplicaInputBuilder() .setShardName("cars").setMemberName("member-3").setDataStoreType(DataStoreType.Config).build()) .get(10, TimeUnit.SECONDS); verifySuccessfulRpcResult(rpcResult); verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2"); verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1"); verifyNoShardPresent(replicaNode3.configDataStore(), "cars"); Cluster.get(leaderNode1.kit().getSystem()).down(Cluster.get(replicaNode2.kit().getSystem()).selfAddress()); replicaNode2.cleanup(); MemberNode newPeplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); newPeplicaNode2.configDataStore().waitTillReady(); verifyRaftPeersPresent(newPeplicaNode2.configDataStore(), "cars", "member-1"); ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(), leaderNode1.operDataStore(), null); rpcResult = service1.removeShardReplica(new RemoveShardReplicaInputBuilder().setShardName("cars") .setMemberName("member-2").setDataStoreType(DataStoreType.Config).build()).get(10, TimeUnit.SECONDS); verifySuccessfulRpcResult(rpcResult); verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars"); verifyNoShardPresent(newPeplicaNode2.configDataStore(), "cars"); }
@Test public void testRemoveShardLeaderReplica() throws Exception { String name = "testRemoveShardLeaderReplica"; String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; final MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder( DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)) .build(); final MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); final MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); leaderNode1.configDataStore().waitTillReady(); verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3"); verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3"); verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2"); replicaNode2.waitForMembersUp("member-1", "member-3"); replicaNode3.waitForMembersUp("member-1", "member-2"); ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(), leaderNode1.operDataStore(), null); RpcResult<RemoveShardReplicaOutput> rpcResult = service1.removeShardReplica(new RemoveShardReplicaInputBuilder() .setShardName("cars").setMemberName("member-1").setDataStoreType(DataStoreType.Config).build()) .get(10, TimeUnit.SECONDS); verifySuccessfulRpcResult(rpcResult); verifyRaftState(replicaNode2.configDataStore(), "cars", raftState -> assertThat("Leader Id", raftState.getLeader(), anyOf(containsString("member-2"), containsString("member-3")))); verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-3"); verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-2"); verifyNoShardPresent(leaderNode1.configDataStore(), "cars"); }