opendaylight/md-sal/sal-cluster-admin-impl/src/main/java/org/opendaylight/controller/cluster/datastore/admin/ClusterAdminRpcService.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore.admin;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.ActorSelection;
  12 import akka.actor.Status.Success;
  13 import akka.dispatch.OnComplete;
  14 import akka.pattern.Patterns;
  15 import akka.util.Timeout;
  16 import com.google.common.base.Strings;
  17 import com.google.common.base.Throwables;
  18 import com.google.common.collect.ImmutableMap;
  19 import com.google.common.util.concurrent.FutureCallback;
  20 import com.google.common.util.concurrent.Futures;
  21 import com.google.common.util.concurrent.ListenableFuture;
  22 import com.google.common.util.concurrent.MoreExecutors;
  23 import com.google.common.util.concurrent.SettableFuture;
  24 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  25 import java.io.FileOutputStream;
  26 import java.io.IOException;
  27 import java.util.AbstractMap.SimpleEntry;
  28 import java.util.ArrayList;
  29 import java.util.HashMap;
  30 import java.util.List;
  31 import java.util.Map;
  32 import java.util.Map.Entry;
  33 import java.util.Set;
  34 import java.util.concurrent.ExecutionException;
  35 import java.util.concurrent.TimeUnit;
  36 import java.util.function.Function;
  37 import java.util.stream.Collectors;
  38 import org.apache.commons.lang3.SerializationUtils;
  39 import org.eclipse.jdt.annotation.NonNull;
  40 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  41 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
  42 import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica;
  43 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
  44 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
  45 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
  46 import org.opendaylight.controller.cluster.datastore.messages.GetKnownClients;
  47 import org.opendaylight.controller.cluster.datastore.messages.GetKnownClientsReply;
  48 import org.opendaylight.controller.cluster.datastore.messages.GetShardRole;
  49 import org.opendaylight.controller.cluster.datastore.messages.GetShardRoleReply;
  50 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
  51 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
  52 import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
  53 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
  54 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
  55 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
  56 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
  57 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  58 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
  59 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaInput;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutput;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutputBuilder;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsInput;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutput;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutputBuilder;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutput;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutputBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutput;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutputBuilder;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsInput;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsInput;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutput;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutputBuilder;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleInput;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutput;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutputBuilder;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleInput;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardInput;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutput;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutputBuilder;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutput;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutputBuilder;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaInput;
 102 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutput;
 103 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutputBuilder;
 104 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
 105 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
 106 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
 107 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.get.known.clients._for.all.shards.output.ShardResult1;
 108 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.get.known.clients._for.all.shards.output.ShardResult1Builder;
 109 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.get.known.clients._for.all.shards.output.shard.result.KnownClientsBuilder;
 110 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LeaderActorRefBuilder;
 111 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LocalBuilder;
 112 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
 113 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
 114 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
 115 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 116 import org.opendaylight.yangtools.yang.common.Empty;
 117 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
 118 import org.opendaylight.yangtools.yang.common.RpcResult;
 119 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 120 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 121 import org.slf4j.Logger;
 122 import org.slf4j.LoggerFactory;
 123
 124 /**
 125  * Implements the yang RPCs defined in the generated ClusterAdminService interface.
 126  *
 127  * @author Thomas Pantelis
 128  */
 129 public class ClusterAdminRpcService implements ClusterAdminService {
 130     private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
 131
 132     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
 133     private static final @NonNull RpcResult<LocateShardOutput> LOCAL_SHARD_RESULT =
 134             RpcResultBuilder.success(new LocateShardOutputBuilder()
 135                 .setMemberNode(new LocalBuilder().setLocal(Empty.getInstance()).build())
 136                 .build())
 137             .build();
 138
 139     private final DistributedDataStoreInterface configDataStore;
 140     private final DistributedDataStoreInterface operDataStore;
 141     private final BindingNormalizedNodeSerializer serializer;
 142     private final Timeout makeLeaderLocalTimeout;
 143
 144     public ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
 145             final DistributedDataStoreInterface operDataStore,
 146             final BindingNormalizedNodeSerializer serializer) {
 147         this.configDataStore = configDataStore;
 148         this.operDataStore = operDataStore;
 149         this.serializer = serializer;
 150
 151         this.makeLeaderLocalTimeout =
 152                 new Timeout(configDataStore.getActorUtils().getDatastoreContext()
 153                         .getShardLeaderElectionTimeout().duration().$times(2));
 154     }
 155
 156     @Override
 157     public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input) {
 158         final String shardName = input.getShardName();
 159         if (Strings.isNullOrEmpty(shardName)) {
 160             return newFailedRpcResultFuture("A valid shard name must be specified");
 161         }
 162
 163         DataStoreType dataStoreType = input.getDataStoreType();
 164         if (dataStoreType == null) {
 165             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 166         }
 167
 168         LOG.info("Adding replica for shard {}", shardName);
 169
 170         final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
 171         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
 172         Futures.addCallback(future, new FutureCallback<Success>() {
 173             @Override
 174             public void onSuccess(final Success success) {
 175                 LOG.info("Successfully added replica for shard {}", shardName);
 176                 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
 177             }
 178
 179             @Override
 180             public void onFailure(final Throwable failure) {
 181                 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
 182                         returnFuture, failure);
 183             }
 184         }, MoreExecutors.directExecutor());
 185
 186         return returnFuture;
 187     }
 188
 189     @Override
 190     public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
 191             final RemoveShardReplicaInput input) {
 192         final String shardName = input.getShardName();
 193         if (Strings.isNullOrEmpty(shardName)) {
 194             return newFailedRpcResultFuture("A valid shard name must be specified");
 195         }
 196
 197         DataStoreType dataStoreType = input.getDataStoreType();
 198         if (dataStoreType == null) {
 199             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 200         }
 201
 202         final String memberName = input.getMemberName();
 203         if (Strings.isNullOrEmpty(memberName)) {
 204             return newFailedRpcResultFuture("A valid member name must be specified");
 205         }
 206
 207         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
 208
 209         final SettableFuture<RpcResult<RemoveShardReplicaOutput>> returnFuture = SettableFuture.create();
 210         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 211                 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
 212         Futures.addCallback(future, new FutureCallback<Success>() {
 213             @Override
 214             public void onSuccess(final Success success) {
 215                 LOG.info("Successfully removed replica for shard {}", shardName);
 216                 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
 217             }
 218
 219             @Override
 220             public void onFailure(final Throwable failure) {
 221                 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
 222                         returnFuture, failure);
 223             }
 224         }, MoreExecutors.directExecutor());
 225
 226         return returnFuture;
 227     }
 228
 229     @Override
 230     public ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input) {
 231         final ActorUtils utils;
 232         switch (input.getDataStoreType()) {
 233             case Config:
 234                 utils = configDataStore.getActorUtils();
 235                 break;
 236             case Operational:
 237                 utils = operDataStore.getActorUtils();
 238                 break;
 239             default:
 240                 return newFailedRpcResultFuture("Unhandled datastore in " + input);
 241         }
 242
 243         final SettableFuture<RpcResult<LocateShardOutput>> ret = SettableFuture.create();
 244         utils.findPrimaryShardAsync(input.getShardName()).onComplete(new OnComplete<PrimaryShardInfo>() {
 245             @Override
 246             public void onComplete(final Throwable failure, final PrimaryShardInfo success) throws Throwable {
 247                 if (failure != null) {
 248                     LOG.debug("Failed to find shard for {}", input, failure);
 249                     ret.setException(failure);
 250                     return;
 251                 }
 252
 253                 // Data tree implies local leak
 254                 if (success.getLocalShardDataTree().isPresent()) {
 255                     ret.set(LOCAL_SHARD_RESULT);
 256                     return;
 257                 }
 258
 259                 final ActorSelection actorPath = success.getPrimaryShardActor();
 260                 ret.set(newSuccessfulResult(new LocateShardOutputBuilder()
 261                     .setMemberNode(new LeaderActorRefBuilder()
 262                         .setLeaderActorRef(actorPath.toSerializationFormat())
 263                         .build())
 264                     .build()));
 265             }
 266         }, utils.getClientDispatcher());
 267
 268         return ret;
 269     }
 270
 271     @Override
 272     public ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input) {
 273         final String shardName = input.getShardName();
 274         if (Strings.isNullOrEmpty(shardName)) {
 275             return newFailedRpcResultFuture("A valid shard name must be specified");
 276         }
 277
 278         DataStoreType dataStoreType = input.getDataStoreType();
 279         if (dataStoreType == null) {
 280             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 281         }
 282
 283         ActorUtils actorUtils = dataStoreType == DataStoreType.Config
 284                 ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
 285
 286         LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
 287                 actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
 288
 289         final scala.concurrent.Future<ActorRef> localShardReply =
 290                 actorUtils.findLocalShardAsync(shardName);
 291
 292         final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
 293         localShardReply.onComplete(new OnComplete<ActorRef>() {
 294             @Override
 295             public void onComplete(final Throwable failure, final ActorRef actorRef) {
 296                 if (failure != null) {
 297                     LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
 298                             + " local shard.", shardName, dataStoreType, failure);
 299                     makeLeaderLocalAsk.failure(failure);
 300                 } else {
 301                     makeLeaderLocalAsk
 302                             .completeWith(actorUtils
 303                                     .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
 304                 }
 305             }
 306         }, actorUtils.getClientDispatcher());
 307
 308         final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
 309         makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
 310             @Override
 311             public void onComplete(final Throwable failure, final Object success) {
 312                 if (failure != null) {
 313                     LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
 314                     future.set(RpcResultBuilder.<MakeLeaderLocalOutput>failed().withError(ErrorType.APPLICATION,
 315                             "leadership transfer failed", failure).build());
 316                     return;
 317                 }
 318
 319                 LOG.debug("Leadership transfer complete");
 320                 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
 321             }
 322         }, actorUtils.getClientDispatcher());
 323
 324         return future;
 325     }
 326
 327     @Override
 328     public ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
 329             final AddPrefixShardReplicaInput input) {
 330
 331         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 332         if (identifier == null) {
 333             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 334         }
 335
 336         final DataStoreType dataStoreType = input.getDataStoreType();
 337         if (dataStoreType == null) {
 338             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 339         }
 340
 341         LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
 342
 343         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 344         final SettableFuture<RpcResult<AddPrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
 345         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
 346         Futures.addCallback(future, new FutureCallback<Success>() {
 347             @Override
 348             public void onSuccess(final Success success) {
 349                 LOG.info("Successfully added replica for shard {}", prefix);
 350                 returnFuture.set(newSuccessfulResult(new AddPrefixShardReplicaOutputBuilder().build()));
 351             }
 352
 353             @Override
 354             public void onFailure(final Throwable failure) {
 355                 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
 356                         returnFuture, failure);
 357             }
 358         }, MoreExecutors.directExecutor());
 359
 360         return returnFuture;
 361     }
 362
 363     @Override
 364     public ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
 365             final RemovePrefixShardReplicaInput input) {
 366
 367         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 368         if (identifier == null) {
 369             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 370         }
 371
 372         final DataStoreType dataStoreType = input.getDataStoreType();
 373         if (dataStoreType == null) {
 374             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 375         }
 376
 377         final String memberName = input.getMemberName();
 378         if (Strings.isNullOrEmpty(memberName)) {
 379             return newFailedRpcResultFuture("A valid member name must be specified");
 380         }
 381
 382         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
 383                 identifier, memberName, dataStoreType);
 384         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 385
 386         final SettableFuture<RpcResult<RemovePrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
 387         final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 388                 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
 389         Futures.addCallback(future, new FutureCallback<Success>() {
 390             @Override
 391             public void onSuccess(final Success success) {
 392                 LOG.info("Successfully removed replica for shard {}", prefix);
 393                 returnFuture.set(newSuccessfulResult(new RemovePrefixShardReplicaOutputBuilder().build()));
 394             }
 395
 396             @Override
 397             public void onFailure(final Throwable failure) {
 398                 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
 399                         returnFuture, failure);
 400             }
 401         }, MoreExecutors.directExecutor());
 402
 403         return returnFuture;
 404     }
 405
 406     @Override
 407     public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
 408             final AddReplicasForAllShardsInput input) {
 409         LOG.info("Adding replicas for all shards");
 410
 411         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 412
 413         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
 414         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
 415
 416         return waitForShardResults(shardResultData, shardResults ->
 417                 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 418                 "Failed to add replica");
 419     }
 420
 421
 422     @Override
 423     public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
 424             final RemoveAllShardReplicasInput input) {
 425         LOG.info("Removing replicas for all shards");
 426
 427         final String memberName = input.getMemberName();
 428         if (Strings.isNullOrEmpty(memberName)) {
 429             return newFailedRpcResultFuture("A valid member name must be specified");
 430         }
 431
 432         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 433         Function<String, Object> messageSupplier = shardName ->
 434                 new RemoveShardReplica(shardName, MemberName.forName(memberName));
 435
 436         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 437         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 438
 439         return waitForShardResults(shardResultData, shardResults ->
 440                 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
 441         "       Failed to remove replica");
 442     }
 443
 444     @Override
 445     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
 446             final ChangeMemberVotingStatesForShardInput input) {
 447         final String shardName = input.getShardName();
 448         if (Strings.isNullOrEmpty(shardName)) {
 449             return newFailedRpcResultFuture("A valid shard name must be specified");
 450         }
 451
 452         DataStoreType dataStoreType = input.getDataStoreType();
 453         if (dataStoreType == null) {
 454             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 455         }
 456
 457         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 458         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 459             return newFailedRpcResultFuture("No member voting state input was specified");
 460         }
 461
 462         ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
 463                 memberVotingStates);
 464
 465         LOG.info("Change member voting states for shard {}: {}", shardName,
 466                 changeVotingStatus.getMeberVotingStatusMap());
 467
 468         final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
 469         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
 470         Futures.addCallback(future, new FutureCallback<Success>() {
 471             @Override
 472             public void onSuccess(final Success success) {
 473                 LOG.info("Successfully changed member voting states for shard {}", shardName);
 474                 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
 475             }
 476
 477             @Override
 478             public void onFailure(final Throwable failure) {
 479                 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
 480                         returnFuture, failure);
 481             }
 482         }, MoreExecutors.directExecutor());
 483
 484         return returnFuture;
 485     }
 486
 487     @Override
 488     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
 489             final ChangeMemberVotingStatesForAllShardsInput input) {
 490         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 491         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 492             return newFailedRpcResultFuture("No member voting state input was specified");
 493         }
 494
 495         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 496         Function<String, Object> messageSupplier = shardName ->
 497                 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
 498
 499         LOG.info("Change member voting states for all shards");
 500
 501         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 502         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 503
 504         return waitForShardResults(shardResultData, shardResults ->
 505                 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 506                 "Failed to change member voting states");
 507     }
 508
 509     @Override
 510     public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
 511             final FlipMemberVotingStatesForAllShardsInput input) {
 512         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 513         Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
 514
 515         LOG.info("Flip member voting states for all shards");
 516
 517         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 518         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 519
 520         return waitForShardResults(shardResultData, shardResults ->
 521                 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 522                 "Failed to change member voting states");
 523     }
 524
 525     @Override
 526     public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
 527         final String shardName = input.getShardName();
 528         if (Strings.isNullOrEmpty(shardName)) {
 529             return newFailedRpcResultFuture("A valid shard name must be specified");
 530         }
 531
 532         DataStoreType dataStoreType = input.getDataStoreType();
 533         if (dataStoreType == null) {
 534             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 535         }
 536
 537         LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
 538
 539         final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
 540         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
 541                 new GetShardRole(shardName));
 542         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
 543             @Override
 544             public void onSuccess(final GetShardRoleReply reply) {
 545                 if (reply == null) {
 546                     returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 547                             "No Shard role present. Please retry..").build());
 548                     return;
 549                 }
 550                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
 551                 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
 552                 if (reply.getRole() != null) {
 553                     builder.setRole(reply.getRole());
 554                 }
 555                 returnFuture.set(newSuccessfulResult(builder.build()));
 556             }
 557
 558             @Override
 559             public void onFailure(final Throwable failure) {
 560                 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 561                         "Failed to get shard role.", failure).build());
 562             }
 563         }, MoreExecutors.directExecutor());
 564
 565         return returnFuture;
 566     }
 567
 568     @Override
 569     public ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
 570             final GetPrefixShardRoleInput input) {
 571         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 572         if (identifier == null) {
 573             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 574         }
 575
 576         final DataStoreType dataStoreType = input.getDataStoreType();
 577         if (dataStoreType == null) {
 578             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 579         }
 580
 581         LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
 582
 583         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 584         final String shardName = ClusterUtils.getCleanShardName(prefix);
 585         final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
 586         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
 587                 new GetShardRole(shardName));
 588         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
 589             @Override
 590             public void onSuccess(final GetShardRoleReply reply) {
 591                 if (reply == null) {
 592                     returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 593                             "No Shard role present. Please retry..").build());
 594                     return;
 595                 }
 596
 597                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
 598                 final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
 599                 if (reply.getRole() != null) {
 600                     builder.setRole(reply.getRole());
 601                 }
 602                 returnFuture.set(newSuccessfulResult(builder.build()));
 603             }
 604
 605             @Override
 606             public void onFailure(final Throwable failure) {
 607                 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 608                         "Failed to get shard role.", failure).build());
 609             }
 610         }, MoreExecutors.directExecutor());
 611
 612         return returnFuture;
 613     }
 614
 615     @Override
 616     public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
 617         LOG.debug("backupDatastore: {}", input);
 618
 619         if (Strings.isNullOrEmpty(input.getFilePath())) {
 620             return newFailedRpcResultFuture("A valid file path must be specified");
 621         }
 622
 623         final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
 624         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
 625         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
 626             @Override
 627             public void onSuccess(final List<DatastoreSnapshot> snapshots) {
 628                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
 629             }
 630
 631             @Override
 632             public void onFailure(final Throwable failure) {
 633                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
 634             }
 635         }, MoreExecutors.directExecutor());
 636
 637         return returnFuture;
 638     }
 639
 640
 641     @Override
 642     public ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
 643             final GetKnownClientsForAllShardsInput input) {
 644         final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies =
 645                 getAllShardLeadersClients();
 646         return Futures.whenAllComplete(allShardReplies.values()).call(() -> processReplies(allShardReplies),
 647             MoreExecutors.directExecutor());
 648     }
 649
 650     private static RpcResult<GetKnownClientsForAllShardsOutput> processReplies(
 651             final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies) {
 652         final List<ShardResult> result = new ArrayList<>(allShardReplies.size());
 653         for (Entry<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> entry : allShardReplies.entrySet()) {
 654             final ListenableFuture<GetKnownClientsReply> future = entry.getValue();
 655             final ShardResultBuilder builder = new ShardResultBuilder()
 656                     .setDataStoreType(entry.getKey().getDataStoreType())
 657                     .setShardName(entry.getKey().getShardName());
 658
 659             final GetKnownClientsReply reply;
 660             try {
 661                 reply = Futures.getDone(future);
 662             } catch (ExecutionException e) {
 663                 LOG.debug("Shard {} failed to answer", entry.getKey(), e);
 664                 result.add(builder.setSucceeded(Boolean.FALSE).setErrorMessage(e.getCause().getMessage()).build());
 665                 continue;
 666             }
 667
 668             result.add(builder
 669                 .setSucceeded(Boolean.TRUE)
 670                 .addAugmentation(ShardResult1.class, new ShardResult1Builder()
 671                     .setKnownClients(reply.getClients().stream()
 672                         .map(client -> new KnownClientsBuilder()
 673                             .setMember(client.getFrontendId().getMemberName().toYang())
 674                             .setType(client.getFrontendId().getClientType().toYang())
 675                             .setGeneration(client.getYangGeneration())
 676                             .build())
 677                         .collect(Collectors.toList()))
 678                     .build())
 679                 .build());
 680         }
 681
 682         return RpcResultBuilder.success(new GetKnownClientsForAllShardsOutputBuilder().setShardResult(result).build())
 683                 .build();
 684     }
 685
 686     private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
 687             final List<MemberVotingState> memberVotingStatus) {
 688         Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
 689         for (MemberVotingState memberStatus: memberVotingStatus) {
 690             serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
 691         }
 692         return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
 693     }
 694
 695     private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
 696             final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
 697             final Function<List<ShardResult>, T> resultDataSupplier,
 698             final String failureLogMsgPrefix) {
 699         final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
 700         final List<ShardResult> shardResults = new ArrayList<>();
 701         for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
 702             Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
 703                 @Override
 704                 public void onSuccess(final Success result) {
 705                     synchronized (shardResults) {
 706                         ShardResultBuilder shardResult = entry.getValue();
 707                         LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
 708                                 shardResult.getDataStoreType());
 709                         shardResults.add(shardResult.setSucceeded(true).build());
 710                         checkIfComplete();
 711                     }
 712                 }
 713
 714                 @Override
 715                 public void onFailure(final Throwable failure) {
 716                     synchronized (shardResults) {
 717                         ShardResultBuilder shardResult = entry.getValue();
 718                         LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
 719                                 shardResult.getDataStoreType(), failure);
 720                         shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
 721                                 Throwables.getRootCause(failure).getMessage()).build());
 722                         checkIfComplete();
 723                     }
 724                 }
 725
 726                 void checkIfComplete() {
 727                     LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
 728                     if (shardResults.size() == shardResultData.size()) {
 729                         returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
 730                     }
 731                 }
 732             }, MoreExecutors.directExecutor());
 733         }
 734         return returnFuture;
 735     }
 736
 737     private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
 738             final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
 739             final Function<String, Object> messageSupplier) {
 740         ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
 741                 : operDataStore.getActorUtils();
 742         Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
 743
 744         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
 745
 746         for (String shardName: allShardNames) {
 747             ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
 748                                                   SHARD_MGR_TIMEOUT);
 749             shardResultData.add(new SimpleEntry<>(future,
 750                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
 751         }
 752     }
 753
 754     private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
 755         Timeout timeout = SHARD_MGR_TIMEOUT;
 756         ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
 757         ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
 758
 759         return Futures.allAsList(configFuture, operFuture);
 760     }
 761
 762     private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
 763         ActorRef shardManager = dataStoreType == DataStoreType.Config
 764                 ? configDataStore.getActorUtils().getShardManager()
 765                         : operDataStore.getActorUtils().getShardManager();
 766         return ask(shardManager, message, SHARD_MGR_TIMEOUT);
 767     }
 768
 769     @SuppressWarnings("checkstyle:IllegalCatch")
 770     private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
 771             final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
 772         try (FileOutputStream fos = new FileOutputStream(fileName)) {
 773             SerializationUtils.serialize(snapshots, fos);
 774
 775             returnFuture.set(newSuccessfulResult(new BackupDatastoreOutputBuilder().build()));
 776             LOG.info("Successfully backed up datastore to file {}", fileName);
 777         } catch (IOException | RuntimeException e) {
 778             onDatastoreBackupFailure(fileName, returnFuture, e);
 779         }
 780     }
 781
 782     private static <T> void onDatastoreBackupFailure(final String fileName,
 783             final SettableFuture<RpcResult<T>> returnFuture, final Throwable failure) {
 784         onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
 785     }
 786
 787     @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
 788     private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
 789             final Throwable failure) {
 790         LOG.error("{}", msg, failure);
 791         returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
 792                 failure.getMessage())).build());
 793     }
 794
 795     private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
 796         final SettableFuture<T> returnFuture = SettableFuture.create();
 797
 798         @SuppressWarnings("unchecked")
 799         scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
 800         askFuture.onComplete(new OnComplete<T>() {
 801             @Override
 802             public void onComplete(final Throwable failure, final T resp) {
 803                 if (failure != null) {
 804                     returnFuture.setException(failure);
 805                 } else {
 806                     returnFuture.set(resp);
 807                 }
 808             }
 809         }, configDataStore.getActorUtils().getClientDispatcher());
 810
 811         return returnFuture;
 812     }
 813
 814     private ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> getAllShardLeadersClients() {
 815         final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder =
 816                 ImmutableMap.builder();
 817
 818         addAllShardsClients(builder, DataStoreType.Config, configDataStore.getActorUtils());
 819         addAllShardsClients(builder, DataStoreType.Operational, operDataStore.getActorUtils());
 820
 821         return builder.build();
 822     }
 823
 824     private static void addAllShardsClients(
 825             final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder,
 826             final DataStoreType type, final ActorUtils utils) {
 827         for (String shardName : utils.getConfiguration().getAllShardNames()) {
 828             final SettableFuture<GetKnownClientsReply> future = SettableFuture.create();
 829             builder.put(new ShardIdentifier(type, shardName), future);
 830
 831             utils.findPrimaryShardAsync(shardName).flatMap(
 832                 info -> Patterns.ask(info.getPrimaryShardActor(), GetKnownClients.INSTANCE, SHARD_MGR_TIMEOUT),
 833                 utils.getClientDispatcher()).onComplete(new OnComplete<Object>() {
 834                     @Override
 835                     public void onComplete(final Throwable failure, final Object success) {
 836                         if (failure == null) {
 837                             future.set((GetKnownClientsReply) success);
 838                         } else {
 839                             future.setException(failure);
 840                         }
 841                     }
 842                 }, utils.getClientDispatcher());
 843         }
 844     }
 845
 846     private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
 847         return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
 848     }
 849
 850     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
 851         return newFailedRpcResultBuilder(message, null);
 852     }
 853
 854     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
 855         return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
 856     }
 857
 858     private static <T> RpcResult<T> newSuccessfulResult(final T data) {
 859         return RpcResultBuilder.success(data).build();
 860     }
 861 }