ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard( final ChangeMemberVotingStatesForShardInput 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"); } List<MemberVotingState> memberVotingStates = input.getMemberVotingState(); if (memberVotingStates == null || memberVotingStates.isEmpty()) { return newFailedRpcResultFuture("No member voting state input was specified"); } ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName, memberVotingStates); LOG.info("Change member voting states for shard {}: {}", shardName, changeVotingStatus.getMeberVotingStatusMap()); final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create(); ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus); Futures.addCallback(future, new FutureCallback<Success>() { @Override public void onSuccess(final Success success) { LOG.info("Successfully changed member voting states for shard {}", shardName); returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build())); } @Override public void onFailure(final Throwable failure) { onMessageFailure(String.format("Failed to change member voting states 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 testChangeMemberVotingStatesForShard() throws Exception { String name = "testChangeMemberVotingStatusForShard"; 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<ChangeMemberVotingStatesForShardOutput> rpcResult = service3 .changeMemberVotingStatesForShard(new ChangeMemberVotingStatesForShardInputBuilder() .setShardName("cars").setDataStoreType(DataStoreType.Config) .setMemberVotingState(ImmutableList.of( new MemberVotingStateBuilder().setMemberName("member-2").setVoting(FALSE).build(), new MemberVotingStateBuilder().setMemberName("member-3").setVoting(FALSE).build())) .build()) .get(10, TimeUnit.SECONDS); verifySuccessfulRpcResult(rpcResult); verifyVotingStates(leaderNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", TRUE), new SimpleEntry<>("member-2", FALSE), new SimpleEntry<>("member-3", FALSE)); verifyVotingStates(replicaNode2.configDataStore(), "cars", new SimpleEntry<>("member-1", TRUE), new SimpleEntry<>("member-2", FALSE), new SimpleEntry<>("member-3", FALSE)); verifyVotingStates(replicaNode3.configDataStore(), "cars", new SimpleEntry<>("member-1", TRUE), new SimpleEntry<>("member-2", FALSE), new SimpleEntry<>("member-3", FALSE)); }
@Test public void testChangeMemberVotingStatesForSingleNodeShard() throws Exception { String name = "testChangeMemberVotingStatesForSingleNodeShard"; String moduleShardsConfig = "module-shards-member1.conf"; MemberNode leaderNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder( DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)) .build(); leaderNode.configDataStore().waitTillReady(); ClusterAdminRpcService service = new ClusterAdminRpcService(leaderNode.configDataStore(), leaderNode.operDataStore(), null); RpcResult<ChangeMemberVotingStatesForShardOutput> rpcResult = service .changeMemberVotingStatesForShard(new ChangeMemberVotingStatesForShardInputBuilder() .setShardName("cars").setDataStoreType(DataStoreType.Config) .setMemberVotingState(ImmutableList .of(new MemberVotingStateBuilder().setMemberName("member-1").setVoting(FALSE).build())) .build()) .get(10, TimeUnit.SECONDS); verifyFailedRpcResult(rpcResult); verifyVotingStates(leaderNode.configDataStore(), "cars", new SimpleEntry<>("member-1", TRUE)); }