class 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;
    }

    public  ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
            final DistributedDataStoreInterface operDataStore,
            final BindingNormalizedNodeSerializer serializer);

}