class ClusterAdminRpcService implements ClusterAdminService {

    @Override
    public ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
            final GetPrefixShardRoleInput input) {
        final InstanceIdentifier<?> identifier = input.getShardPrefix();
        if (identifier == null) {
            return newFailedRpcResultFuture("A valid shard identifier must be specified");
        }

        final DataStoreType dataStoreType = input.getDataStoreType();
        if (dataStoreType == null) {
            return newFailedRpcResultFuture("A valid DataStoreType must be specified");
        }

        LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);

        final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
        final String shardName = ClusterUtils.getCleanShardName(prefix);
        final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
        ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
                new GetShardRole(shardName));
        Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
            @Override
            public void onSuccess(final GetShardRoleReply reply) {
                if (reply == null) {
                    returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
                            "No Shard role present. Please retry..").build());
                    return;
                }

                LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
                final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
                if (reply.getRole() != null) {
                    builder.setRole(reply.getRole());
                }
                returnFuture.set(newSuccessfulResult(builder.build()));
            }

            @Override
            public void onFailure(final Throwable failure) {
                returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
                        "Failed to get shard role.", failure).build());
            }
        }, MoreExecutors.directExecutor());

        return returnFuture;
    }

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

}