opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java

   1 /*
   2  * Copyright (c) 2014 Cisco 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
   9 package org.opendaylight.controller.cluster.datastore.shardmanager;
  10
  11 import static akka.pattern.Patterns.ask;
  12 import akka.actor.ActorRef;
  13 import akka.actor.Address;
  14 import akka.actor.Cancellable;
  15 import akka.actor.OneForOneStrategy;
  16 import akka.actor.PoisonPill;
  17 import akka.actor.Props;
  18 import akka.actor.Status;
  19 import akka.actor.SupervisorStrategy;
  20 import akka.cluster.ClusterEvent;
  21 import akka.dispatch.Futures;
  22 import akka.dispatch.OnComplete;
  23 import akka.japi.Function;
  24 import akka.pattern.Patterns;
  25 import akka.persistence.RecoveryCompleted;
  26 import akka.persistence.SaveSnapshotFailure;
  27 import akka.persistence.SaveSnapshotSuccess;
  28 import akka.persistence.SnapshotOffer;
  29 import akka.persistence.SnapshotSelectionCriteria;
  30 import akka.serialization.Serialization;
  31 import akka.util.Timeout;
  32 import com.google.common.annotations.VisibleForTesting;
  33 import com.google.common.base.Objects;
  34 import com.google.common.base.Optional;
  35 import com.google.common.base.Preconditions;
  36 import com.google.common.base.Strings;
  37 import com.google.common.base.Supplier;
  38 import com.google.common.collect.Sets;
  39 import java.io.ByteArrayInputStream;
  40 import java.io.ObjectInputStream;
  41 import java.util.ArrayList;
  42 import java.util.Collection;
  43 import java.util.Collections;
  44 import java.util.HashMap;
  45 import java.util.HashSet;
  46 import java.util.Iterator;
  47 import java.util.List;
  48 import java.util.Map;
  49 import java.util.Set;
  50 import java.util.concurrent.CountDownLatch;
  51 import java.util.concurrent.TimeUnit;
  52 import java.util.concurrent.TimeoutException;
  53 import javax.annotation.Nonnull;
  54 import javax.annotation.Nullable;
  55 import org.apache.commons.lang3.SerializationUtils;
  56 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
  57 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
  58 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  59 import org.opendaylight.controller.cluster.datastore.DatastoreContextFactory;
  60 import org.opendaylight.controller.cluster.datastore.Shard;
  61 import org.opendaylight.controller.cluster.datastore.config.Configuration;
  62 import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
  63 import org.opendaylight.controller.cluster.datastore.exceptions.AlreadyExistsException;
  64 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  65 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
  66 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
  67 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  68 import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
  69 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
  70 import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
  71 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
  72 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  73 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
  74 import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
  75 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  76 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
  77 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
  78 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
  79 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
  80 import org.opendaylight.controller.cluster.datastore.messages.RemoteFindPrimary;
  81 import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
  82 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
  83 import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
  84 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  85 import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
  86 import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
  87 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
  88 import org.opendaylight.controller.cluster.notifications.RoleChangeNotification;
  89 import org.opendaylight.controller.cluster.raft.RaftState;
  90 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
  91 import org.opendaylight.controller.cluster.raft.base.messages.SwitchBehavior;
  92 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
  93 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
  94 import org.opendaylight.controller.cluster.raft.messages.AddServer;
  95 import org.opendaylight.controller.cluster.raft.messages.AddServerReply;
  96 import org.opendaylight.controller.cluster.raft.messages.RemoveServer;
  97 import org.opendaylight.controller.cluster.raft.messages.RemoveServerReply;
  98 import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
  99 import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
 100 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
 101 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
 102 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 103 import org.slf4j.Logger;
 104 import org.slf4j.LoggerFactory;
 105 import scala.concurrent.ExecutionContext;
 106 import scala.concurrent.Future;
 107 import scala.concurrent.duration.Duration;
 108 import scala.concurrent.duration.FiniteDuration;
 109
 110 /**
 111  * The ShardManager has the following jobs,
 112  * <ul>
 113  * <li> Create all the local shard replicas that belong on this cluster member
 114  * <li> Find the address of the local shard
 115  * <li> Find the primary replica for any given shard
 116  * <li> Monitor the cluster members and store their addresses
 117  * <ul>
 118  */
 119 class ShardManager extends AbstractUntypedPersistentActorWithMetering {
 120
 121     private static final Logger LOG = LoggerFactory.getLogger(ShardManager.class);
 122
 123     // Stores a mapping between a shard name and it's corresponding information
 124     // Shard names look like inventory, topology etc and are as specified in
 125     // configuration
 126     private final Map<String, ShardInformation> localShards = new HashMap<>();
 127
 128     // The type of a ShardManager reflects the type of the datastore itself
 129     // A data store could be of type config/operational
 130     private final String type;
 131
 132     private final ClusterWrapper cluster;
 133
 134     private final Configuration configuration;
 135
 136     private final String shardDispatcherPath;
 137
 138     private final ShardManagerInfo mBean;
 139
 140     private DatastoreContextFactory datastoreContextFactory;
 141
 142     private final CountDownLatch waitTillReadyCountdownLatch;
 143
 144     private final PrimaryShardInfoFutureCache primaryShardInfoCache;
 145
 146     private final ShardPeerAddressResolver peerAddressResolver;
 147
 148     private SchemaContext schemaContext;
 149
 150     private DatastoreSnapshot restoreFromSnapshot;
 151
 152     private ShardManagerSnapshot currentSnapshot;
 153
 154     private final Set<String> shardReplicaOperationsInProgress = new HashSet<>();
 155
 156     private final String persistenceId;
 157
 158     ShardManager(AbstractShardManagerCreator<?> builder) {
 159         this.cluster = builder.getCluster();
 160         this.configuration = builder.getConfiguration();
 161         this.datastoreContextFactory = builder.getDdatastoreContextFactory();
 162         this.type = datastoreContextFactory.getBaseDatastoreContext().getDataStoreName();
 163         this.shardDispatcherPath =
 164                 new Dispatchers(context().system().dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard);
 165         this.waitTillReadyCountdownLatch = builder.getWaitTillReadyCountdownLatch();
 166         this.primaryShardInfoCache = builder.getPrimaryShardInfoCache();
 167         this.restoreFromSnapshot = builder.getRestoreFromSnapshot();
 168
 169         String possiblePersistenceId = datastoreContextFactory.getBaseDatastoreContext().getShardManagerPersistenceId();
 170         persistenceId = possiblePersistenceId != null ? possiblePersistenceId : "shard-manager-" + type;
 171
 172         peerAddressResolver = new ShardPeerAddressResolver(type, cluster.getCurrentMemberName());
 173
 174         // Subscribe this actor to cluster member events
 175         cluster.subscribeToMemberEvents(getSelf());
 176
 177         mBean = new ShardManagerInfo(getSelf(), cluster.getCurrentMemberName(), "shard-manager-" + this.type,
 178                 datastoreContextFactory.getBaseDatastoreContext().getDataStoreMXBeanType());
 179         mBean.registerMBean();
 180     }
 181
 182     @Override
 183     public void postStop() {
 184         LOG.info("Stopping ShardManager {}", persistenceId());
 185
 186         mBean.unregisterMBean();
 187     }
 188
 189     @Override
 190     public void handleCommand(Object message) throws Exception {
 191         if (message  instanceof FindPrimary) {
 192             findPrimary((FindPrimary)message);
 193         } else if(message instanceof FindLocalShard){
 194             findLocalShard((FindLocalShard) message);
 195         } else if (message instanceof UpdateSchemaContext) {
 196             updateSchemaContext(message);
 197         } else if(message instanceof ActorInitialized) {
 198             onActorInitialized(message);
 199         } else if (message instanceof ClusterEvent.MemberUp){
 200             memberUp((ClusterEvent.MemberUp) message);
 201         } else if (message instanceof ClusterEvent.MemberExited){
 202             memberExited((ClusterEvent.MemberExited) message);
 203         } else if(message instanceof ClusterEvent.MemberRemoved) {
 204             memberRemoved((ClusterEvent.MemberRemoved) message);
 205         } else if(message instanceof ClusterEvent.UnreachableMember) {
 206             memberUnreachable((ClusterEvent.UnreachableMember)message);
 207         } else if(message instanceof ClusterEvent.ReachableMember) {
 208             memberReachable((ClusterEvent.ReachableMember) message);
 209         } else if(message instanceof DatastoreContextFactory) {
 210             onDatastoreContextFactory((DatastoreContextFactory)message);
 211         } else if(message instanceof RoleChangeNotification) {
 212             onRoleChangeNotification((RoleChangeNotification) message);
 213         } else if(message instanceof FollowerInitialSyncUpStatus){
 214             onFollowerInitialSyncStatus((FollowerInitialSyncUpStatus) message);
 215         } else if(message instanceof ShardNotInitializedTimeout) {
 216             onShardNotInitializedTimeout((ShardNotInitializedTimeout)message);
 217         } else if(message instanceof ShardLeaderStateChanged) {
 218             onLeaderStateChanged((ShardLeaderStateChanged) message);
 219         } else if(message instanceof SwitchShardBehavior){
 220             onSwitchShardBehavior((SwitchShardBehavior) message);
 221         } else if(message instanceof CreateShard) {
 222             onCreateShard((CreateShard)message);
 223         } else if(message instanceof AddShardReplica){
 224             onAddShardReplica((AddShardReplica)message);
 225         } else if(message instanceof ForwardedAddServerReply) {
 226             ForwardedAddServerReply msg = (ForwardedAddServerReply)message;
 227             onAddServerReply(msg.shardInfo, msg.addServerReply, getSender(), msg.leaderPath,
 228                     msg.removeShardOnFailure);
 229         } else if(message instanceof ForwardedAddServerFailure) {
 230             ForwardedAddServerFailure msg = (ForwardedAddServerFailure)message;
 231             onAddServerFailure(msg.shardName, msg.failureMessage, msg.failure, getSender(), msg.removeShardOnFailure);
 232         } else if(message instanceof PrimaryShardFoundForContext) {
 233             PrimaryShardFoundForContext primaryShardFoundContext = (PrimaryShardFoundForContext)message;
 234             onPrimaryShardFoundContext(primaryShardFoundContext);
 235         } else if(message instanceof RemoveShardReplica) {
 236             onRemoveShardReplica((RemoveShardReplica) message);
 237         } else if(message instanceof WrappedShardResponse){
 238             onWrappedShardResponse((WrappedShardResponse) message);
 239         } else if(message instanceof GetSnapshot) {
 240             onGetSnapshot();
 241         } else if(message instanceof ServerRemoved){
 242             onShardReplicaRemoved((ServerRemoved) message);
 243         } else if(message instanceof SaveSnapshotSuccess) {
 244             onSaveSnapshotSuccess((SaveSnapshotSuccess)message);
 245         } else if(message instanceof SaveSnapshotFailure) {
 246             LOG.error("{}: SaveSnapshotFailure received for saving snapshot of shards",
 247                     persistenceId(), ((SaveSnapshotFailure) message).cause());
 248         } else if(message instanceof Shutdown) {
 249             onShutDown();
 250         } else if (message instanceof GetLocalShardIds) {
 251             onGetLocalShardIds();
 252         } else {
 253             unknownMessage(message);
 254         }
 255     }
 256
 257     private void onShutDown() {
 258         List<Future<Boolean>> stopFutures = new ArrayList<>(localShards.size());
 259         for (ShardInformation info : localShards.values()) {
 260             if (info.getActor() != null) {
 261                 LOG.debug("{}: Issuing gracefulStop to shard {}", persistenceId(), info.getShardId());
 262
 263                 FiniteDuration duration = info.getDatastoreContext().getShardRaftConfig().getElectionTimeOutInterval().$times(2);
 264                 stopFutures.add(Patterns.gracefulStop(info.getActor(), duration, Shutdown.INSTANCE));
 265             }
 266         }
 267
 268         LOG.info("Shutting down ShardManager {} - waiting on {} shards", persistenceId(), stopFutures.size());
 269
 270         ExecutionContext dispatcher = new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client);
 271         Future<Iterable<Boolean>> combinedFutures = Futures.sequence(stopFutures, dispatcher);
 272
 273         combinedFutures.onComplete(new OnComplete<Iterable<Boolean>>() {
 274             @Override
 275             public void onComplete(Throwable failure, Iterable<Boolean> results) {
 276                 LOG.debug("{}: All shards shutdown - sending PoisonPill to self", persistenceId());
 277
 278                 self().tell(PoisonPill.getInstance(), self());
 279
 280                 if(failure != null) {
 281                     LOG.warn("{}: An error occurred attempting to shut down the shards", persistenceId(), failure);
 282                 } else {
 283                     int nfailed = 0;
 284                     for(Boolean r: results) {
 285                         if(!r) {
 286                             nfailed++;
 287                         }
 288                     }
 289
 290                     if(nfailed > 0) {
 291                         LOG.warn("{}: {} shards did not shut down gracefully", persistenceId(), nfailed);
 292                     }
 293                 }
 294             }
 295         }, dispatcher);
 296     }
 297
 298     private void onWrappedShardResponse(WrappedShardResponse message) {
 299         if (message.getResponse() instanceof RemoveServerReply) {
 300             onRemoveServerReply(getSender(), message.getShardId(), (RemoveServerReply) message.getResponse(),
 301                     message.getLeaderPath());
 302         }
 303     }
 304
 305     private void onRemoveServerReply(ActorRef originalSender, ShardIdentifier shardId, RemoveServerReply replyMsg,
 306             String leaderPath) {
 307         shardReplicaOperationsInProgress.remove(shardId);
 308
 309         LOG.debug ("{}: Received {} for shard {}", persistenceId(), replyMsg, shardId.getShardName());
 310
 311         if (replyMsg.getStatus() == ServerChangeStatus.OK) {
 312             LOG.debug ("{}: Leader shard successfully removed the replica shard {}", persistenceId(),
 313                     shardId.getShardName());
 314             originalSender.tell(new Status.Success(null), getSelf());
 315         } else {
 316             LOG.warn ("{}: Leader failed to remove shard replica {} with status {}",
 317                     persistenceId(), shardId, replyMsg.getStatus());
 318
 319             Exception failure = getServerChangeException(RemoveServer.class, replyMsg.getStatus(), leaderPath, shardId);
 320             originalSender.tell(new Status.Failure(failure), getSelf());
 321         }
 322     }
 323
 324     private void onPrimaryShardFoundContext(PrimaryShardFoundForContext primaryShardFoundContext) {
 325         if(primaryShardFoundContext.getContextMessage() instanceof AddShardReplica) {
 326             addShard(primaryShardFoundContext.getShardName(), primaryShardFoundContext.getRemotePrimaryShardFound(),
 327                     getSender());
 328         } else if(primaryShardFoundContext.getContextMessage() instanceof RemoveShardReplica){
 329             removeShardReplica((RemoveShardReplica) primaryShardFoundContext.getContextMessage(),
 330                     primaryShardFoundContext.getShardName(), primaryShardFoundContext.getPrimaryPath(), getSender());
 331         }
 332     }
 333
 334     private void removeShardReplica(RemoveShardReplica contextMessage, final String shardName, final String primaryPath,
 335             final ActorRef sender) {
 336         if(isShardReplicaOperationInProgress(shardName, sender)) {
 337             return;
 338         }
 339
 340         shardReplicaOperationsInProgress.add(shardName);
 341
 342         final ShardIdentifier shardId = getShardIdentifier(contextMessage.getMemberName(), shardName);
 343
 344         final DatastoreContext datastoreContext = newShardDatastoreContextBuilder(shardName).build();
 345
 346         //inform ShardLeader to remove this shard as a replica by sending an RemoveServer message
 347         LOG.debug ("{}: Sending RemoveServer message to peer {} for shard {}", persistenceId(),
 348                 primaryPath, shardId);
 349
 350         Timeout removeServerTimeout = new Timeout(datastoreContext.getShardLeaderElectionTimeout().
 351                 duration());
 352         Future<Object> futureObj = ask(getContext().actorSelection(primaryPath),
 353                 new RemoveServer(shardId.toString()), removeServerTimeout);
 354
 355         futureObj.onComplete(new OnComplete<Object>() {
 356             @Override
 357             public void onComplete(Throwable failure, Object response) {
 358                 if (failure != null) {
 359                     String msg = String.format("RemoveServer request to leader %s for shard %s failed",
 360                             primaryPath, shardName);
 361
 362                     LOG.debug ("{}: {}", persistenceId(), msg, failure);
 363
 364                     // FAILURE
 365                     sender.tell(new Status.Failure(new RuntimeException(msg, failure)), self());
 366                 } else {
 367                     // SUCCESS
 368                     self().tell(new WrappedShardResponse(shardId, response, primaryPath), sender);
 369                 }
 370             }
 371         }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client));
 372     }
 373
 374     private void onShardReplicaRemoved(ServerRemoved message) {
 375         final ShardIdentifier shardId = new ShardIdentifier.Builder().fromShardIdString(message.getServerId()).build();
 376         final ShardInformation shardInformation = localShards.remove(shardId.getShardName());
 377         if(shardInformation == null) {
 378             LOG.debug("{} : Shard replica {} is not present in list", persistenceId(), shardId.toString());
 379             return;
 380         } else if(shardInformation.getActor() != null) {
 381             LOG.debug("{} : Sending Shutdown to Shard actor {}", persistenceId(), shardInformation.getActor());
 382             shardInformation.getActor().tell(Shutdown.INSTANCE, self());
 383         }
 384         LOG.debug("{} : Local Shard replica for shard {} has been removed", persistenceId(), shardId.getShardName());
 385         persistShardList();
 386     }
 387
 388     private void onGetSnapshot() {
 389         LOG.debug("{}: onGetSnapshot", persistenceId());
 390
 391         List<String> notInitialized = null;
 392         for(ShardInformation shardInfo: localShards.values()) {
 393             if(!shardInfo.isShardInitialized()) {
 394                 if(notInitialized == null) {
 395                     notInitialized = new ArrayList<>();
 396                 }
 397
 398                 notInitialized.add(shardInfo.getShardName());
 399             }
 400         }
 401
 402         if(notInitialized != null) {
 403             getSender().tell(new Status.Failure(new IllegalStateException(String.format(
 404                     "%d shard(s) %s are not initialized", notInitialized.size(), notInitialized))), getSelf());
 405             return;
 406         }
 407
 408         byte[] shardManagerSnapshot = null;
 409         if(currentSnapshot != null) {
 410             shardManagerSnapshot = SerializationUtils.serialize(currentSnapshot);
 411         }
 412
 413         ActorRef replyActor = getContext().actorOf(ShardManagerGetSnapshotReplyActor.props(
 414                 new ArrayList<>(localShards.keySet()), type, shardManagerSnapshot , getSender(), persistenceId(),
 415                 datastoreContextFactory.getBaseDatastoreContext().getShardInitializationTimeout().duration()));
 416
 417         for(ShardInformation shardInfo: localShards.values()) {
 418             shardInfo.getActor().tell(GetSnapshot.INSTANCE, replyActor);
 419         }
 420     }
 421
 422     private void onCreateShard(CreateShard createShard) {
 423         LOG.debug("{}: onCreateShard: {}", persistenceId(), createShard);
 424
 425         Object reply;
 426         try {
 427             String shardName = createShard.getModuleShardConfig().getShardName();
 428             if(localShards.containsKey(shardName)) {
 429                 LOG.debug("{}: Shard {} already exists", persistenceId(), shardName);
 430                 reply = new Status.Success(String.format("Shard with name %s already exists", shardName));
 431             } else {
 432                 doCreateShard(createShard);
 433                 reply = new Status.Success(null);
 434             }
 435         } catch (Exception e) {
 436             LOG.error("{}: onCreateShard failed", persistenceId(), e);
 437             reply = new Status.Failure(e);
 438         }
 439
 440         if(getSender() != null && !getContext().system().deadLetters().equals(getSender())) {
 441             getSender().tell(reply, getSelf());
 442         }
 443     }
 444
 445     private void doCreateShard(CreateShard createShard) {
 446         ModuleShardConfiguration moduleShardConfig = createShard.getModuleShardConfig();
 447         String shardName = moduleShardConfig.getShardName();
 448
 449         configuration.addModuleShardConfiguration(moduleShardConfig);
 450
 451         DatastoreContext shardDatastoreContext = createShard.getDatastoreContext();
 452         if(shardDatastoreContext == null) {
 453             shardDatastoreContext = newShardDatastoreContext(shardName);
 454         } else {
 455             shardDatastoreContext = DatastoreContext.newBuilderFrom(shardDatastoreContext).shardPeerAddressResolver(
 456                     peerAddressResolver).build();
 457         }
 458
 459         ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), shardName);
 460
 461         boolean shardWasInRecoveredSnapshot = currentSnapshot != null &&
 462                 currentSnapshot.getShardList().contains(shardName);
 463
 464         Map<String, String> peerAddresses;
 465         boolean isActiveMember;
 466         if(shardWasInRecoveredSnapshot || configuration.getMembersFromShardName(shardName).
 467                 contains(cluster.getCurrentMemberName())) {
 468             peerAddresses = getPeerAddresses(shardName);
 469             isActiveMember = true;
 470         } else {
 471             // The local member is not in the static shard member configuration and the shard did not
 472             // previously exist (ie !shardWasInRecoveredSnapshot). In this case we'll create
 473             // the shard with no peers and with elections disabled so it stays as follower. A
 474             // subsequent AddServer request will be needed to make it an active member.
 475             isActiveMember = false;
 476             peerAddresses = Collections.emptyMap();
 477             shardDatastoreContext = DatastoreContext.newBuilderFrom(shardDatastoreContext).
 478                     customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()).build();
 479         }
 480
 481         LOG.debug("{} doCreateShard: shardId: {}, memberNames: {}, peerAddresses: {}, isActiveMember: {}",
 482                 persistenceId(), shardId, moduleShardConfig.getShardMemberNames(), peerAddresses,
 483                 isActiveMember);
 484
 485         ShardInformation info = new ShardInformation(shardName, shardId, peerAddresses,
 486                 shardDatastoreContext, createShard.getShardBuilder(), peerAddressResolver);
 487         info.setActiveMember(isActiveMember);
 488         localShards.put(info.getShardName(), info);
 489
 490         if(schemaContext != null) {
 491             info.setActor(newShardActor(schemaContext, info));
 492         }
 493     }
 494
 495     private DatastoreContext.Builder newShardDatastoreContextBuilder(String shardName) {
 496         return DatastoreContext.newBuilderFrom(datastoreContextFactory.getShardDatastoreContext(shardName)).
 497                 shardPeerAddressResolver(peerAddressResolver);
 498     }
 499
 500     private DatastoreContext newShardDatastoreContext(String shardName) {
 501         return newShardDatastoreContextBuilder(shardName).build();
 502     }
 503
 504     private void checkReady(){
 505         if (isReadyWithLeaderId()) {
 506             LOG.info("{}: All Shards are ready - data store {} is ready, available count is {}",
 507                     persistenceId(), type, waitTillReadyCountdownLatch.getCount());
 508
 509             waitTillReadyCountdownLatch.countDown();
 510         }
 511     }
 512
 513     private void onLeaderStateChanged(ShardLeaderStateChanged leaderStateChanged) {
 514         LOG.info("{}: Received LeaderStateChanged message: {}", persistenceId(), leaderStateChanged);
 515
 516         ShardInformation shardInformation = findShardInformation(leaderStateChanged.getMemberId());
 517         if(shardInformation != null) {
 518             shardInformation.setLocalDataTree(leaderStateChanged.getLocalShardDataTree());
 519             shardInformation.setLeaderVersion(leaderStateChanged.getLeaderPayloadVersion());
 520             if(shardInformation.setLeaderId(leaderStateChanged.getLeaderId())) {
 521                 primaryShardInfoCache.remove(shardInformation.getShardName());
 522             }
 523
 524             checkReady();
 525         } else {
 526             LOG.debug("No shard found with member Id {}", leaderStateChanged.getMemberId());
 527         }
 528     }
 529
 530     private void onShardNotInitializedTimeout(ShardNotInitializedTimeout message) {
 531         ShardInformation shardInfo = message.getShardInfo();
 532
 533         LOG.debug("{}: Received ShardNotInitializedTimeout message for shard {}", persistenceId(),
 534                 shardInfo.getShardName());
 535
 536         shardInfo.removeOnShardInitialized(message.getOnShardInitialized());
 537
 538         if(!shardInfo.isShardInitialized()) {
 539             LOG.debug("{}: Returning NotInitializedException for shard {}", persistenceId(), shardInfo.getShardName());
 540             message.getSender().tell(createNotInitializedException(shardInfo.shardId), getSelf());
 541         } else {
 542             LOG.debug("{}: Returning NoShardLeaderException for shard {}", persistenceId(), shardInfo.getShardName());
 543             message.getSender().tell(createNoShardLeaderException(shardInfo.shardId), getSelf());
 544         }
 545     }
 546
 547     private void onFollowerInitialSyncStatus(FollowerInitialSyncUpStatus status) {
 548         LOG.info("{} Received follower initial sync status for {} status sync done {}", persistenceId(),
 549                 status.getName(), status.isInitialSyncDone());
 550
 551         ShardInformation shardInformation = findShardInformation(status.getName());
 552
 553         if(shardInformation != null) {
 554             shardInformation.setFollowerSyncStatus(status.isInitialSyncDone());
 555
 556             mBean.setSyncStatus(isInSync());
 557         }
 558
 559     }
 560
 561     private void onRoleChangeNotification(RoleChangeNotification roleChanged) {
 562         LOG.info("{}: Received role changed for {} from {} to {}", persistenceId(), roleChanged.getMemberId(),
 563                 roleChanged.getOldRole(), roleChanged.getNewRole());
 564
 565         ShardInformation shardInformation = findShardInformation(roleChanged.getMemberId());
 566         if(shardInformation != null) {
 567             shardInformation.setRole(roleChanged.getNewRole());
 568             checkReady();
 569             mBean.setSyncStatus(isInSync());
 570         }
 571     }
 572
 573
 574     private ShardInformation findShardInformation(String memberId) {
 575         for(ShardInformation info : localShards.values()){
 576             if(info.getShardId().toString().equals(memberId)){
 577                 return info;
 578             }
 579         }
 580
 581         return null;
 582     }
 583
 584     private boolean isReadyWithLeaderId() {
 585         boolean isReady = true;
 586         for (ShardInformation info : localShards.values()) {
 587             if(!info.isShardReadyWithLeaderId()){
 588                 isReady = false;
 589                 break;
 590             }
 591         }
 592         return isReady;
 593     }
 594
 595     private boolean isInSync(){
 596         for (ShardInformation info : localShards.values()) {
 597             if(!info.isInSync()){
 598                 return false;
 599             }
 600         }
 601         return true;
 602     }
 603
 604     private void onActorInitialized(Object message) {
 605         final ActorRef sender = getSender();
 606
 607         if (sender == null) {
 608             return; //why is a non-actor sending this message? Just ignore.
 609         }
 610
 611         String actorName = sender.path().name();
 612         //find shard name from actor name; actor name is stringified shardId
 613         ShardIdentifier shardId = ShardIdentifier.builder().fromShardIdString(actorName).build();
 614
 615         if (shardId.getShardName() == null) {
 616             return;
 617         }
 618
 619         markShardAsInitialized(shardId.getShardName());
 620     }
 621
 622     private void markShardAsInitialized(String shardName) {
 623         LOG.debug("{}: Initializing shard [{}]", persistenceId(), shardName);
 624
 625         ShardInformation shardInformation = localShards.get(shardName);
 626         if (shardInformation != null) {
 627             shardInformation.setActorInitialized();
 628
 629             shardInformation.getActor().tell(new RegisterRoleChangeListener(), self());
 630         }
 631     }
 632
 633     @Override
 634     protected void handleRecover(Object message) throws Exception {
 635         if (message instanceof RecoveryCompleted) {
 636             onRecoveryCompleted();
 637         } else if (message instanceof SnapshotOffer) {
 638             applyShardManagerSnapshot((ShardManagerSnapshot)((SnapshotOffer) message).snapshot());
 639         }
 640     }
 641
 642     private void onRecoveryCompleted() {
 643         LOG.info("Recovery complete : {}", persistenceId());
 644
 645         // We no longer persist SchemaContext modules so delete all the prior messages from the akka
 646         // journal on upgrade from Helium.
 647         deleteMessages(lastSequenceNr());
 648
 649         if(currentSnapshot == null && restoreFromSnapshot != null &&
 650                 restoreFromSnapshot.getShardManagerSnapshot() != null) {
 651             try(ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(
 652                     restoreFromSnapshot.getShardManagerSnapshot()))) {
 653                 ShardManagerSnapshot snapshot = (ShardManagerSnapshot) ois.readObject();
 654
 655                 LOG.debug("{}: Deserialized restored ShardManagerSnapshot: {}", persistenceId(), snapshot);
 656
 657                 applyShardManagerSnapshot(snapshot);
 658             } catch(Exception e) {
 659                 LOG.error("{}: Error deserializing restored ShardManagerSnapshot", persistenceId(), e);
 660             }
 661         }
 662
 663         createLocalShards();
 664     }
 665
 666     private void findLocalShard(FindLocalShard message) {
 667         final ShardInformation shardInformation = localShards.get(message.getShardName());
 668
 669         if(shardInformation == null){
 670             getSender().tell(new LocalShardNotFound(message.getShardName()), getSelf());
 671             return;
 672         }
 673
 674         sendResponse(shardInformation, message.isWaitUntilInitialized(), false, new Supplier<Object>() {
 675             @Override
 676             public Object get() {
 677                 return new LocalShardFound(shardInformation.getActor());
 678             }
 679         });
 680     }
 681
 682     private void sendResponse(ShardInformation shardInformation, boolean doWait,
 683             boolean wantShardReady, final Supplier<Object> messageSupplier) {
 684         if (!shardInformation.isShardInitialized() || (wantShardReady && !shardInformation.isShardReadyWithLeaderId())) {
 685             if(doWait) {
 686                 final ActorRef sender = getSender();
 687                 final ActorRef self = self();
 688
 689                 Runnable replyRunnable = new Runnable() {
 690                     @Override
 691                     public void run() {
 692                         sender.tell(messageSupplier.get(), self);
 693                     }
 694                 };
 695
 696                 OnShardInitialized onShardInitialized = wantShardReady ? new OnShardReady(replyRunnable) :
 697                     new OnShardInitialized(replyRunnable);
 698
 699                 shardInformation.addOnShardInitialized(onShardInitialized);
 700
 701                 FiniteDuration timeout = shardInformation.getDatastoreContext().getShardInitializationTimeout().duration();
 702                 if(shardInformation.isShardInitialized()) {
 703                     // If the shard is already initialized then we'll wait enough time for the shard to
 704                     // elect a leader, ie 2 times the election timeout.
 705                     timeout = FiniteDuration.create(shardInformation.getDatastoreContext().getShardRaftConfig()
 706                             .getElectionTimeOutInterval().toMillis() * 2, TimeUnit.MILLISECONDS);
 707                 }
 708
 709                 LOG.debug("{}: Scheduling {} ms timer to wait for shard {}", persistenceId(), timeout.toMillis(),
 710                         shardInformation.getShardName());
 711
 712                 Cancellable timeoutSchedule = getContext().system().scheduler().scheduleOnce(
 713                         timeout, getSelf(),
 714                         new ShardNotInitializedTimeout(shardInformation, onShardInitialized, sender),
 715                         getContext().dispatcher(), getSelf());
 716
 717                 onShardInitialized.setTimeoutSchedule(timeoutSchedule);
 718
 719             } else if (!shardInformation.isShardInitialized()) {
 720                 LOG.debug("{}: Returning NotInitializedException for shard {}", persistenceId(),
 721                         shardInformation.getShardName());
 722                 getSender().tell(createNotInitializedException(shardInformation.shardId), getSelf());
 723             } else {
 724                 LOG.debug("{}: Returning NoShardLeaderException for shard {}", persistenceId(),
 725                         shardInformation.getShardName());
 726                 getSender().tell(createNoShardLeaderException(shardInformation.shardId), getSelf());
 727             }
 728
 729             return;
 730         }
 731
 732         getSender().tell(messageSupplier.get(), getSelf());
 733     }
 734
 735     private static NoShardLeaderException createNoShardLeaderException(ShardIdentifier shardId) {
 736         return new NoShardLeaderException(null, shardId.toString());
 737     }
 738
 739     private static NotInitializedException createNotInitializedException(ShardIdentifier shardId) {
 740         return new NotInitializedException(String.format(
 741                 "Found primary shard %s but it's not initialized yet. Please try again later", shardId));
 742     }
 743
 744     private void memberRemoved(ClusterEvent.MemberRemoved message) {
 745         String memberName = message.member().roles().iterator().next();
 746
 747         LOG.debug("{}: Received MemberRemoved: memberName: {}, address: {}", persistenceId(), memberName,
 748                 message.member().address());
 749
 750         peerAddressResolver.removePeerAddress(memberName);
 751
 752         for(ShardInformation info : localShards.values()){
 753             info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
 754         }
 755     }
 756
 757     private void memberExited(ClusterEvent.MemberExited message) {
 758         String memberName = message.member().roles().iterator().next();
 759
 760         LOG.debug("{}: Received MemberExited: memberName: {}, address: {}", persistenceId(), memberName,
 761                 message.member().address());
 762
 763         peerAddressResolver.removePeerAddress(memberName);
 764
 765         for(ShardInformation info : localShards.values()){
 766             info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
 767         }
 768     }
 769
 770     private void memberUp(ClusterEvent.MemberUp message) {
 771         String memberName = message.member().roles().iterator().next();
 772
 773         LOG.debug("{}: Received MemberUp: memberName: {}, address: {}", persistenceId(), memberName,
 774                 message.member().address());
 775
 776         addPeerAddress(memberName, message.member().address());
 777
 778         checkReady();
 779     }
 780
 781     private void addPeerAddress(String memberName, Address address) {
 782         peerAddressResolver.addPeerAddress(memberName, address);
 783
 784         for(ShardInformation info : localShards.values()){
 785             String shardName = info.getShardName();
 786             String peerId = getShardIdentifier(memberName, shardName).toString();
 787             info.updatePeerAddress(peerId, peerAddressResolver.getShardActorAddress(shardName, memberName), getSelf());
 788
 789             info.peerUp(memberName, peerId, getSelf());
 790         }
 791     }
 792
 793     private void memberReachable(ClusterEvent.ReachableMember message) {
 794         String memberName = message.member().roles().iterator().next();
 795         LOG.debug("Received ReachableMember: memberName {}, address: {}", memberName, message.member().address());
 796
 797         addPeerAddress(memberName, message.member().address());
 798
 799         markMemberAvailable(memberName);
 800     }
 801
 802     private void memberUnreachable(ClusterEvent.UnreachableMember message) {
 803         String memberName = message.member().roles().iterator().next();
 804         LOG.debug("Received UnreachableMember: memberName {}, address: {}", memberName, message.member().address());
 805
 806         markMemberUnavailable(memberName);
 807     }
 808
 809     private void markMemberUnavailable(final String memberName) {
 810         for(ShardInformation info : localShards.values()){
 811             String leaderId = info.getLeaderId();
 812             if(leaderId != null && leaderId.contains(memberName)) {
 813                 LOG.debug("Marking Leader {} as unavailable.", leaderId);
 814                 info.setLeaderAvailable(false);
 815
 816                 primaryShardInfoCache.remove(info.getShardName());
 817             }
 818
 819             info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
 820         }
 821     }
 822
 823     private void markMemberAvailable(final String memberName) {
 824         for(ShardInformation info : localShards.values()){
 825             String leaderId = info.getLeaderId();
 826             if(leaderId != null && leaderId.contains(memberName)) {
 827                 LOG.debug("Marking Leader {} as available.", leaderId);
 828                 info.setLeaderAvailable(true);
 829             }
 830
 831             info.peerUp(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
 832         }
 833     }
 834
 835     private void onDatastoreContextFactory(DatastoreContextFactory factory) {
 836         datastoreContextFactory = factory;
 837         for (ShardInformation info : localShards.values()) {
 838             info.setDatastoreContext(newShardDatastoreContext(info.getShardName()), getSelf());
 839         }
 840     }
 841
 842     private void onGetLocalShardIds() {
 843         final List<String> response = new ArrayList<>(localShards.size());
 844
 845         for (ShardInformation info : localShards.values()) {
 846             response.add(info.getShardId().toString());
 847         }
 848
 849         getSender().tell(new Status.Success(response), getSelf());
 850     }
 851
 852     private void onSwitchShardBehavior(final SwitchShardBehavior message) {
 853         final ShardIdentifier identifier = message.getShardId();
 854
 855         if (identifier != null) {
 856             final ShardInformation info = localShards.get(identifier.getShardName());
 857             if (info == null) {
 858                 getSender().tell(new Status.Failure(
 859                     new IllegalArgumentException("Shard " + identifier + " is not local")), getSelf());
 860                 return;
 861             }
 862
 863             switchShardBehavior(info, new SwitchBehavior(message.getNewState(), message.getTerm()));
 864         } else {
 865             for (ShardInformation info : localShards.values()) {
 866                 switchShardBehavior(info, new SwitchBehavior(message.getNewState(), message.getTerm()));
 867             }
 868         }
 869
 870         getSender().tell(new Status.Success(null), getSelf());
 871     }
 872
 873     private void switchShardBehavior(final ShardInformation info, final SwitchBehavior switchBehavior) {
 874         final ActorRef actor = info.getActor();
 875         if (actor != null) {
 876             actor.tell(switchBehavior, getSelf());
 877           } else {
 878             LOG.warn("Could not switch the behavior of shard {} to {} - shard is not yet available",
 879                 info.shardName, switchBehavior.getNewState());
 880         }
 881     }
 882
 883     /**
 884      * Notifies all the local shards of a change in the schema context
 885      *
 886      * @param message
 887      */
 888     private void updateSchemaContext(final Object message) {
 889         schemaContext = ((UpdateSchemaContext) message).getSchemaContext();
 890
 891         LOG.debug("Got updated SchemaContext: # of modules {}", schemaContext.getAllModuleIdentifiers().size());
 892
 893         for (ShardInformation info : localShards.values()) {
 894             if (info.getActor() == null) {
 895                 LOG.debug("Creating Shard {}", info.getShardId());
 896                 info.setActor(newShardActor(schemaContext, info));
 897             } else {
 898                 info.getActor().tell(message, getSelf());
 899             }
 900         }
 901     }
 902
 903     @VisibleForTesting
 904     protected ClusterWrapper getCluster() {
 905         return cluster;
 906     }
 907
 908     @VisibleForTesting
 909     protected ActorRef newShardActor(final SchemaContext schemaContext, ShardInformation info) {
 910         return getContext().actorOf(info.newProps(schemaContext)
 911                 .withDispatcher(shardDispatcherPath), info.getShardId().toString());
 912     }
 913
 914     private void findPrimary(FindPrimary message) {
 915         LOG.debug("{}: In findPrimary: {}", persistenceId(), message);
 916
 917         final String shardName = message.getShardName();
 918         final boolean canReturnLocalShardState = !(message instanceof RemoteFindPrimary);
 919
 920         // First see if the there is a local replica for the shard
 921         final ShardInformation info = localShards.get(shardName);
 922         if (info != null && info.isActiveMember()) {
 923             sendResponse(info, message.isWaitUntilReady(), true, new Supplier<Object>() {
 924                 @Override
 925                 public Object get() {
 926                     String primaryPath = info.getSerializedLeaderActor();
 927                     Object found = canReturnLocalShardState && info.isLeader() ?
 928                             new LocalPrimaryShardFound(primaryPath, info.getLocalShardDataTree().get()) :
 929                                 new RemotePrimaryShardFound(primaryPath, info.getLeaderVersion());
 930
 931                             if(LOG.isDebugEnabled()) {
 932                                 LOG.debug("{}: Found primary for {}: {}", persistenceId(), shardName, found);
 933                             }
 934
 935                             return found;
 936                 }
 937             });
 938
 939             return;
 940         }
 941
 942         final Collection<String> visitedAddresses;
 943         if(message instanceof RemoteFindPrimary) {
 944             visitedAddresses = ((RemoteFindPrimary)message).getVisitedAddresses();
 945         } else {
 946             visitedAddresses = new ArrayList<>(1);
 947         }
 948
 949         visitedAddresses.add(peerAddressResolver.getShardManagerActorPathBuilder(cluster.getSelfAddress()).toString());
 950
 951         for(String address: peerAddressResolver.getShardManagerPeerActorAddresses()) {
 952             if(visitedAddresses.contains(address)) {
 953                 continue;
 954             }
 955
 956             LOG.debug("{}: findPrimary for {} forwarding to remote ShardManager {}, visitedAddresses: {}",
 957                     persistenceId(), shardName, address, visitedAddresses);
 958
 959             getContext().actorSelection(address).forward(new RemoteFindPrimary(shardName,
 960                     message.isWaitUntilReady(), visitedAddresses), getContext());
 961             return;
 962         }
 963
 964         LOG.debug("{}: No shard found for {}", persistenceId(), shardName);
 965
 966         getSender().tell(new PrimaryNotFoundException(
 967                 String.format("No primary shard found for %s.", shardName)), getSelf());
 968     }
 969
 970     /**
 971      * Construct the name of the shard actor given the name of the member on
 972      * which the shard resides and the name of the shard
 973      *
 974      * @param memberName
 975      * @param shardName
 976      * @return
 977      */
 978     private ShardIdentifier getShardIdentifier(String memberName, String shardName){
 979         return peerAddressResolver.getShardIdentifier(memberName, shardName);
 980     }
 981
 982     /**
 983      * Create shards that are local to the member on which the ShardManager
 984      * runs
 985      *
 986      */
 987     private void createLocalShards() {
 988         String memberName = this.cluster.getCurrentMemberName();
 989         Collection<String> memberShardNames = this.configuration.getMemberShardNames(memberName);
 990
 991         Map<String, DatastoreSnapshot.ShardSnapshot> shardSnapshots = new HashMap<>();
 992         if(restoreFromSnapshot != null)
 993         {
 994             for(DatastoreSnapshot.ShardSnapshot snapshot: restoreFromSnapshot.getShardSnapshots()) {
 995                 shardSnapshots.put(snapshot.getName(), snapshot);
 996             }
 997         }
 998
 999         restoreFromSnapshot = null; // null out to GC
1000
1001         for(String shardName : memberShardNames){
1002             ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
1003
1004             LOG.debug("{}: Creating local shard: {}", persistenceId(), shardId);
1005
1006             Map<String, String> peerAddresses = getPeerAddresses(shardName);
1007             localShards.put(shardName, new ShardInformation(shardName, shardId, peerAddresses,
1008                     newShardDatastoreContext(shardName), Shard.builder().restoreFromSnapshot(
1009                         shardSnapshots.get(shardName)), peerAddressResolver));
1010         }
1011     }
1012
1013     /**
1014      * Given the name of the shard find the addresses of all it's peers
1015      *
1016      * @param shardName
1017      */
1018     private Map<String, String> getPeerAddresses(String shardName) {
1019         Collection<String> members = configuration.getMembersFromShardName(shardName);
1020         Map<String, String> peerAddresses = new HashMap<>();
1021
1022         String currentMemberName = this.cluster.getCurrentMemberName();
1023
1024         for(String memberName : members) {
1025             if(!currentMemberName.equals(memberName)) {
1026                 ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
1027                 String address = peerAddressResolver.getShardActorAddress(shardName, memberName);
1028                 peerAddresses.put(shardId.toString(), address);
1029             }
1030         }
1031         return peerAddresses;
1032     }
1033
1034     @Override
1035     public SupervisorStrategy supervisorStrategy() {
1036
1037         return new OneForOneStrategy(10, Duration.create("1 minute"),
1038                 new Function<Throwable, SupervisorStrategy.Directive>() {
1039             @Override
1040             public SupervisorStrategy.Directive apply(Throwable t) {
1041                 LOG.warn("Supervisor Strategy caught unexpected exception - resuming", t);
1042                 return SupervisorStrategy.resume();
1043             }
1044         }
1045                 );
1046
1047     }
1048
1049     @Override
1050     public String persistenceId() {
1051         return persistenceId;
1052     }
1053
1054     @VisibleForTesting
1055     ShardManagerInfoMBean getMBean(){
1056         return mBean;
1057     }
1058
1059     private boolean isShardReplicaOperationInProgress(final String shardName, final ActorRef sender) {
1060         if (shardReplicaOperationsInProgress.contains(shardName)) {
1061             String msg = String.format("A shard replica operation for %s is already in progress", shardName);
1062             LOG.debug ("{}: {}", persistenceId(), msg);
1063             sender.tell(new Status.Failure(new IllegalStateException(msg)), getSelf());
1064             return true;
1065         }
1066
1067         return false;
1068     }
1069
1070     private void onAddShardReplica (final AddShardReplica shardReplicaMsg) {
1071         final String shardName = shardReplicaMsg.getShardName();
1072
1073         LOG.debug("{}: onAddShardReplica: {}", persistenceId(), shardReplicaMsg);
1074
1075         // verify the shard with the specified name is present in the cluster configuration
1076         if (!(this.configuration.isShardConfigured(shardName))) {
1077             String msg = String.format("No module configuration exists for shard %s", shardName);
1078             LOG.debug ("{}: {}", persistenceId(), msg);
1079             getSender().tell(new Status.Failure(new IllegalArgumentException(msg)), getSelf());
1080             return;
1081         }
1082
1083         // Create the localShard
1084         if (schemaContext == null) {
1085             String msg = String.format(
1086                   "No SchemaContext is available in order to create a local shard instance for %s", shardName);
1087             LOG.debug ("{}: {}", persistenceId(), msg);
1088             getSender().tell(new Status.Failure(new IllegalStateException(msg)), getSelf());
1089             return;
1090         }
1091
1092         findPrimary(shardName, new AutoFindPrimaryFailureResponseHandler(getSender(), shardName, persistenceId(), getSelf()) {
1093             @Override
1094             public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) {
1095                 getSelf().tell(new PrimaryShardFoundForContext(getShardName(), shardReplicaMsg, response), getTargetActor());
1096             }
1097
1098             @Override
1099             public void onLocalPrimaryFound(LocalPrimaryShardFound message) {
1100                 sendLocalReplicaAlreadyExistsReply(getShardName(), getTargetActor());
1101             }
1102
1103         });
1104     }
1105
1106     private void sendLocalReplicaAlreadyExistsReply(String shardName, ActorRef sender) {
1107         String msg = String.format("Local shard %s already exists", shardName);
1108         LOG.debug ("{}: {}", persistenceId(), msg);
1109         sender.tell(new Status.Failure(new AlreadyExistsException(msg)), getSelf());
1110     }
1111
1112     private void addShard(final String shardName, final RemotePrimaryShardFound response, final ActorRef sender) {
1113         if(isShardReplicaOperationInProgress(shardName, sender)) {
1114             return;
1115         }
1116
1117         shardReplicaOperationsInProgress.add(shardName);
1118
1119         final ShardInformation shardInfo;
1120         final boolean removeShardOnFailure;
1121         ShardInformation existingShardInfo = localShards.get(shardName);
1122         if(existingShardInfo == null) {
1123             removeShardOnFailure = true;
1124             ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), shardName);
1125
1126             DatastoreContext datastoreContext = newShardDatastoreContextBuilder(shardName).customRaftPolicyImplementation(
1127                     DisableElectionsRaftPolicy.class.getName()).build();
1128
1129             shardInfo = new ShardInformation(shardName, shardId, getPeerAddresses(shardName), datastoreContext,
1130                     Shard.builder(), peerAddressResolver);
1131             shardInfo.setActiveMember(false);
1132             localShards.put(shardName, shardInfo);
1133             shardInfo.setActor(newShardActor(schemaContext, shardInfo));
1134         } else {
1135             removeShardOnFailure = false;
1136             shardInfo = existingShardInfo;
1137         }
1138
1139         String localShardAddress = peerAddressResolver.getShardActorAddress(shardName, cluster.getCurrentMemberName());
1140
1141         //inform ShardLeader to add this shard as a replica by sending an AddServer message
1142         LOG.debug ("{}: Sending AddServer message to peer {} for shard {}", persistenceId(),
1143                 response.getPrimaryPath(), shardInfo.getShardId());
1144
1145         Timeout addServerTimeout = new Timeout(shardInfo.getDatastoreContext().getShardLeaderElectionTimeout().
1146                 duration());
1147         Future<Object> futureObj = ask(getContext().actorSelection(response.getPrimaryPath()),
1148             new AddServer(shardInfo.getShardId().toString(), localShardAddress, true), addServerTimeout);
1149
1150         futureObj.onComplete(new OnComplete<Object>() {
1151             @Override
1152             public void onComplete(Throwable failure, Object addServerResponse) {
1153                 if (failure != null) {
1154                     LOG.debug ("{}: AddServer request to {} for {} failed", persistenceId(),
1155                             response.getPrimaryPath(), shardName, failure);
1156
1157                     String msg = String.format("AddServer request to leader %s for shard %s failed",
1158                             response.getPrimaryPath(), shardName);
1159                     self().tell(new ForwardedAddServerFailure(shardName, msg, failure, removeShardOnFailure), sender);
1160                 } else {
1161                     self().tell(new ForwardedAddServerReply(shardInfo, (AddServerReply)addServerResponse,
1162                             response.getPrimaryPath(), removeShardOnFailure), sender);
1163                 }
1164             }
1165         }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client));
1166     }
1167
1168     private void onAddServerFailure(String shardName, String message, Throwable failure, ActorRef sender,
1169             boolean removeShardOnFailure) {
1170         shardReplicaOperationsInProgress.remove(shardName);
1171
1172         if(removeShardOnFailure) {
1173             ShardInformation shardInfo = localShards.remove(shardName);
1174             if (shardInfo.getActor() != null) {
1175                 shardInfo.getActor().tell(PoisonPill.getInstance(), getSelf());
1176             }
1177         }
1178
1179         sender.tell(new Status.Failure(message == null ? failure :
1180             new RuntimeException(message, failure)), getSelf());
1181     }
1182
1183     private void onAddServerReply(ShardInformation shardInfo, AddServerReply replyMsg, ActorRef sender,
1184             String leaderPath, boolean removeShardOnFailure) {
1185         String shardName = shardInfo.getShardName();
1186         shardReplicaOperationsInProgress.remove(shardName);
1187
1188         LOG.debug ("{}: Received {} for shard {} from leader {}", persistenceId(), replyMsg, shardName, leaderPath);
1189
1190         if (replyMsg.getStatus() == ServerChangeStatus.OK) {
1191             LOG.debug ("{}: Leader shard successfully added the replica shard {}", persistenceId(), shardName);
1192
1193             // Make the local shard voting capable
1194             shardInfo.setDatastoreContext(newShardDatastoreContext(shardName), getSelf());
1195             shardInfo.setActiveMember(true);
1196             persistShardList();
1197
1198             sender.tell(new Status.Success(null), getSelf());
1199         } else if(replyMsg.getStatus() == ServerChangeStatus.ALREADY_EXISTS) {
1200             sendLocalReplicaAlreadyExistsReply(shardName, sender);
1201         } else {
1202             LOG.warn ("{}: Leader failed to add shard replica {} with status {}",
1203                     persistenceId(), shardName, replyMsg.getStatus());
1204
1205             Exception failure = getServerChangeException(AddServer.class, replyMsg.getStatus(), leaderPath, shardInfo.getShardId());
1206
1207             onAddServerFailure(shardName, null, failure, sender, removeShardOnFailure);
1208         }
1209     }
1210
1211     private static Exception getServerChangeException(Class<?> serverChange, ServerChangeStatus serverChangeStatus,
1212                                                String leaderPath, ShardIdentifier shardId) {
1213         Exception failure;
1214         switch (serverChangeStatus) {
1215             case TIMEOUT:
1216                 failure = new TimeoutException(String.format(
1217                         "The shard leader %s timed out trying to replicate the initial data to the new shard %s." +
1218                         "Possible causes - there was a problem replicating the data or shard leadership changed while replicating the shard data",
1219                         leaderPath, shardId.getShardName()));
1220                 break;
1221             case NO_LEADER:
1222                 failure = createNoShardLeaderException(shardId);
1223                 break;
1224             case NOT_SUPPORTED:
1225                 failure = new UnsupportedOperationException(String.format("%s request is not supported for shard %s",
1226                         serverChange.getSimpleName(), shardId.getShardName()));
1227                 break;
1228             default :
1229                 failure = new RuntimeException(String.format(
1230                         "%s request to leader %s for shard %s failed with status %s",
1231                         serverChange.getSimpleName(), leaderPath, shardId.getShardName(), serverChangeStatus));
1232         }
1233         return failure;
1234     }
1235
1236     private void onRemoveShardReplica (final RemoveShardReplica shardReplicaMsg) {
1237         LOG.debug("{}: onRemoveShardReplica: {}", persistenceId(), shardReplicaMsg);
1238
1239         findPrimary(shardReplicaMsg.getShardName(), new AutoFindPrimaryFailureResponseHandler(getSender(),
1240                 shardReplicaMsg.getShardName(), persistenceId(), getSelf()) {
1241             @Override
1242             public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) {
1243                 getSelf().tell(new PrimaryShardFoundForContext(getShardName(), shardReplicaMsg, response), getTargetActor());
1244             }
1245
1246             @Override
1247             public void onLocalPrimaryFound(LocalPrimaryShardFound response) {
1248                 getSelf().tell(new PrimaryShardFoundForContext(getShardName(), shardReplicaMsg, response), getTargetActor());
1249             }
1250         });
1251     }
1252
1253     private void persistShardList() {
1254         List<String> shardList = new ArrayList<>(localShards.keySet());
1255         for (ShardInformation shardInfo : localShards.values()) {
1256             if (!shardInfo.isActiveMember()) {
1257                 shardList.remove(shardInfo.getShardName());
1258             }
1259         }
1260         LOG.debug ("{}: persisting the shard list {}", persistenceId(), shardList);
1261         saveSnapshot(updateShardManagerSnapshot(shardList));
1262     }
1263
1264     private ShardManagerSnapshot updateShardManagerSnapshot(List<String> shardList) {
1265         currentSnapshot = new ShardManagerSnapshot(shardList);
1266         return currentSnapshot;
1267     }
1268
1269     private void applyShardManagerSnapshot(ShardManagerSnapshot snapshot) {
1270         currentSnapshot = snapshot;
1271
1272         LOG.debug ("{}: onSnapshotOffer: {}", persistenceId(), currentSnapshot);
1273
1274         String currentMember = cluster.getCurrentMemberName();
1275         Set<String> configuredShardList =
1276             new HashSet<>(configuration.getMemberShardNames(currentMember));
1277         for (String shard : currentSnapshot.getShardList()) {
1278             if (!configuredShardList.contains(shard)) {
1279                 // add the current member as a replica for the shard
1280                 LOG.debug ("{}: adding shard {}", persistenceId(), shard);
1281                 configuration.addMemberReplicaForShard(shard, currentMember);
1282             } else {
1283                 configuredShardList.remove(shard);
1284             }
1285         }
1286         for (String shard : configuredShardList) {
1287             // remove the member as a replica for the shard
1288             LOG.debug ("{}: removing shard {}", persistenceId(), shard);
1289             configuration.removeMemberReplicaForShard(shard, currentMember);
1290         }
1291     }
1292
1293     private void onSaveSnapshotSuccess (SaveSnapshotSuccess successMessage) {
1294         LOG.debug ("{} saved ShardManager snapshot successfully. Deleting the prev snapshot if available",
1295             persistenceId());
1296         deleteSnapshots(new SnapshotSelectionCriteria(scala.Long.MaxValue(), successMessage.metadata().timestamp() - 1,
1297             0, 0));
1298     }
1299
1300     private static class ForwardedAddServerReply {
1301         ShardInformation shardInfo;
1302         AddServerReply addServerReply;
1303         String leaderPath;
1304         boolean removeShardOnFailure;
1305
1306         ForwardedAddServerReply(ShardInformation shardInfo, AddServerReply addServerReply, String leaderPath,
1307                 boolean removeShardOnFailure) {
1308             this.shardInfo = shardInfo;
1309             this.addServerReply = addServerReply;
1310             this.leaderPath = leaderPath;
1311             this.removeShardOnFailure = removeShardOnFailure;
1312         }
1313     }
1314
1315     private static class ForwardedAddServerFailure {
1316         String shardName;
1317         String failureMessage;
1318         Throwable failure;
1319         boolean removeShardOnFailure;
1320
1321         ForwardedAddServerFailure(String shardName, String failureMessage, Throwable failure,
1322                 boolean removeShardOnFailure) {
1323             this.shardName = shardName;
1324             this.failureMessage = failureMessage;
1325             this.failure = failure;
1326             this.removeShardOnFailure = removeShardOnFailure;
1327         }
1328     }
1329
1330     @VisibleForTesting
1331     protected static class ShardInformation {
1332         private final ShardIdentifier shardId;
1333         private final String shardName;
1334         private ActorRef actor;
1335         private final Map<String, String> initialPeerAddresses;
1336         private Optional<DataTree> localShardDataTree;
1337         private boolean leaderAvailable = false;
1338
1339         // flag that determines if the actor is ready for business
1340         private boolean actorInitialized = false;
1341
1342         private boolean followerSyncStatus = false;
1343
1344         private final Set<OnShardInitialized> onShardInitializedSet = Sets.newHashSet();
1345         private String role ;
1346         private String leaderId;
1347         private short leaderVersion;
1348
1349         private DatastoreContext datastoreContext;
1350         private Shard.AbstractBuilder<?, ?> builder;
1351         private final ShardPeerAddressResolver addressResolver;
1352         private boolean isActiveMember = true;
1353
1354         private ShardInformation(String shardName, ShardIdentifier shardId,
1355                 Map<String, String> initialPeerAddresses, DatastoreContext datastoreContext,
1356                 Shard.AbstractBuilder<?, ?> builder, ShardPeerAddressResolver addressResolver) {
1357             this.shardName = shardName;
1358             this.shardId = shardId;
1359             this.initialPeerAddresses = initialPeerAddresses;
1360             this.datastoreContext = datastoreContext;
1361             this.builder = builder;
1362             this.addressResolver = addressResolver;
1363         }
1364
1365         Props newProps(SchemaContext schemaContext) {
1366             Preconditions.checkNotNull(builder);
1367             Props props = builder.id(shardId).peerAddresses(initialPeerAddresses).datastoreContext(datastoreContext).
1368                     schemaContext(schemaContext).props();
1369             builder = null;
1370             return props;
1371         }
1372
1373         String getShardName() {
1374             return shardName;
1375         }
1376
1377         @Nullable
1378         ActorRef getActor(){
1379             return actor;
1380         }
1381
1382         void setActor(ActorRef actor) {
1383             this.actor = actor;
1384         }
1385
1386         ShardIdentifier getShardId() {
1387             return shardId;
1388         }
1389
1390         void setLocalDataTree(Optional<DataTree> localShardDataTree) {
1391             this.localShardDataTree = localShardDataTree;
1392         }
1393
1394         Optional<DataTree> getLocalShardDataTree() {
1395             return localShardDataTree;
1396         }
1397
1398         DatastoreContext getDatastoreContext() {
1399             return datastoreContext;
1400         }
1401
1402         void setDatastoreContext(DatastoreContext datastoreContext, ActorRef sender) {
1403             this.datastoreContext = datastoreContext;
1404             if (actor != null) {
1405                 LOG.debug ("Sending new DatastoreContext to {}", shardId);
1406                 actor.tell(this.datastoreContext, sender);
1407             }
1408         }
1409
1410         void updatePeerAddress(String peerId, String peerAddress, ActorRef sender){
1411             LOG.info("updatePeerAddress for peer {} with address {}", peerId, peerAddress);
1412
1413             if(actor != null) {
1414                 if(LOG.isDebugEnabled()) {
1415                     LOG.debug("Sending PeerAddressResolved for peer {} with address {} to {}",
1416                             peerId, peerAddress, actor.path());
1417                 }
1418
1419                 actor.tell(new PeerAddressResolved(peerId, peerAddress), sender);
1420             }
1421
1422             notifyOnShardInitializedCallbacks();
1423         }
1424
1425         void peerDown(String memberName, String peerId, ActorRef sender) {
1426             if(actor != null) {
1427                 actor.tell(new PeerDown(memberName, peerId), sender);
1428             }
1429         }
1430
1431         void peerUp(String memberName, String peerId, ActorRef sender) {
1432             if(actor != null) {
1433                 actor.tell(new PeerUp(memberName, peerId), sender);
1434             }
1435         }
1436
1437         boolean isShardReady() {
1438             return !RaftState.Candidate.name().equals(role) && !Strings.isNullOrEmpty(role);
1439         }
1440
1441         boolean isShardReadyWithLeaderId() {
1442             return leaderAvailable && isShardReady() && !RaftState.IsolatedLeader.name().equals(role) &&
1443                     (isLeader() || addressResolver.resolve(leaderId) != null);
1444         }
1445
1446         boolean isShardInitialized() {
1447             return getActor() != null && actorInitialized;
1448         }
1449
1450         boolean isLeader() {
1451             return Objects.equal(leaderId, shardId.toString());
1452         }
1453
1454         String getSerializedLeaderActor() {
1455             if(isLeader()) {
1456                 return Serialization.serializedActorPath(getActor());
1457             } else {
1458                 return addressResolver.resolve(leaderId);
1459             }
1460         }
1461
1462         void setActorInitialized() {
1463             LOG.debug("Shard {} is initialized", shardId);
1464
1465             this.actorInitialized = true;
1466
1467             notifyOnShardInitializedCallbacks();
1468         }
1469
1470         private void notifyOnShardInitializedCallbacks() {
1471             if(onShardInitializedSet.isEmpty()) {
1472                 return;
1473             }
1474
1475             boolean ready = isShardReadyWithLeaderId();
1476
1477             if(LOG.isDebugEnabled()) {
1478                 LOG.debug("Shard {} is {} - notifying {} OnShardInitialized callbacks", shardId,
1479                         ready ? "ready" : "initialized", onShardInitializedSet.size());
1480             }
1481
1482             Iterator<OnShardInitialized> iter = onShardInitializedSet.iterator();
1483             while(iter.hasNext()) {
1484                 OnShardInitialized onShardInitialized = iter.next();
1485                 if(!(onShardInitialized instanceof OnShardReady) || ready) {
1486                     iter.remove();
1487                     onShardInitialized.getTimeoutSchedule().cancel();
1488                     onShardInitialized.getReplyRunnable().run();
1489                 }
1490             }
1491         }
1492
1493         void addOnShardInitialized(OnShardInitialized onShardInitialized) {
1494             onShardInitializedSet.add(onShardInitialized);
1495         }
1496
1497         void removeOnShardInitialized(OnShardInitialized onShardInitialized) {
1498             onShardInitializedSet.remove(onShardInitialized);
1499         }
1500
1501         void setRole(String newRole) {
1502             this.role = newRole;
1503
1504             notifyOnShardInitializedCallbacks();
1505         }
1506
1507         void setFollowerSyncStatus(boolean syncStatus){
1508             this.followerSyncStatus = syncStatus;
1509         }
1510
1511         boolean isInSync(){
1512             if(RaftState.Follower.name().equals(this.role)){
1513                 return followerSyncStatus;
1514             } else if(RaftState.Leader.name().equals(this.role)){
1515                 return true;
1516             }
1517
1518             return false;
1519         }
1520
1521         boolean setLeaderId(String leaderId) {
1522             boolean changed = !Objects.equal(this.leaderId, leaderId);
1523             this.leaderId = leaderId;
1524             if(leaderId != null) {
1525                 this.leaderAvailable = true;
1526             }
1527             notifyOnShardInitializedCallbacks();
1528
1529             return changed;
1530         }
1531
1532         String getLeaderId() {
1533             return leaderId;
1534         }
1535
1536         void setLeaderAvailable(boolean leaderAvailable) {
1537             this.leaderAvailable = leaderAvailable;
1538
1539             if(leaderAvailable) {
1540                 notifyOnShardInitializedCallbacks();
1541             }
1542         }
1543
1544         short getLeaderVersion() {
1545             return leaderVersion;
1546         }
1547
1548         void setLeaderVersion(short leaderVersion) {
1549             this.leaderVersion = leaderVersion;
1550         }
1551
1552         boolean isActiveMember() {
1553             return isActiveMember;
1554         }
1555
1556         void setActiveMember(boolean isActiveMember) {
1557             this.isActiveMember = isActiveMember;
1558         }
1559     }
1560
1561     private static class OnShardInitialized {
1562         private final Runnable replyRunnable;
1563         private Cancellable timeoutSchedule;
1564
1565         OnShardInitialized(Runnable replyRunnable) {
1566             this.replyRunnable = replyRunnable;
1567         }
1568
1569         Runnable getReplyRunnable() {
1570             return replyRunnable;
1571         }
1572
1573         Cancellable getTimeoutSchedule() {
1574             return timeoutSchedule;
1575         }
1576
1577         void setTimeoutSchedule(Cancellable timeoutSchedule) {
1578             this.timeoutSchedule = timeoutSchedule;
1579         }
1580     }
1581
1582     private static class OnShardReady extends OnShardInitialized {
1583         OnShardReady(Runnable replyRunnable) {
1584             super(replyRunnable);
1585         }
1586     }
1587
1588     private static class ShardNotInitializedTimeout {
1589         private final ActorRef sender;
1590         private final ShardInformation shardInfo;
1591         private final OnShardInitialized onShardInitialized;
1592
1593         ShardNotInitializedTimeout(ShardInformation shardInfo, OnShardInitialized onShardInitialized, ActorRef sender) {
1594             this.sender = sender;
1595             this.shardInfo = shardInfo;
1596             this.onShardInitialized = onShardInitialized;
1597         }
1598
1599         ActorRef getSender() {
1600             return sender;
1601         }
1602
1603         ShardInformation getShardInfo() {
1604             return shardInfo;
1605         }
1606
1607         OnShardInitialized getOnShardInitialized() {
1608             return onShardInitialized;
1609         }
1610     }
1611
1612     private void findPrimary(final String shardName, final FindPrimaryResponseHandler handler) {
1613         Timeout findPrimaryTimeout = new Timeout(datastoreContextFactory.getBaseDatastoreContext().
1614                 getShardInitializationTimeout().duration().$times(2));
1615
1616
1617         Future<Object> futureObj = ask(getSelf(), new FindPrimary(shardName, true), findPrimaryTimeout);
1618         futureObj.onComplete(new OnComplete<Object>() {
1619             @Override
1620             public void onComplete(Throwable failure, Object response) {
1621                 if (failure != null) {
1622                     handler.onFailure(failure);
1623                 } else {
1624                     if(response instanceof RemotePrimaryShardFound) {
1625                         handler.onRemotePrimaryShardFound((RemotePrimaryShardFound) response);
1626                     } else if(response instanceof LocalPrimaryShardFound) {
1627                         handler.onLocalPrimaryFound((LocalPrimaryShardFound) response);
1628                     } else {
1629                         handler.onUnknownResponse(response);
1630                     }
1631                 }
1632             }
1633         }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client));
1634     }
1635
1636     /**
1637      * The FindPrimaryResponseHandler provides specific callback methods which are invoked when a response to the
1638      * a remote or local find primary message is processed
1639      */
1640     private static interface FindPrimaryResponseHandler {
1641         /**
1642          * Invoked when a Failure message is received as a response
1643          *
1644          * @param failure
1645          */
1646         void onFailure(Throwable failure);
1647
1648         /**
1649          * Invoked when a RemotePrimaryShardFound response is received
1650          *
1651          * @param response
1652          */
1653         void onRemotePrimaryShardFound(RemotePrimaryShardFound response);
1654
1655         /**
1656          * Invoked when a LocalPrimaryShardFound response is received
1657          * @param response
1658          */
1659         void onLocalPrimaryFound(LocalPrimaryShardFound response);
1660
1661         /**
1662          * Invoked when an unknown response is received. This is another type of failure.
1663          *
1664          * @param response
1665          */
1666         void onUnknownResponse(Object response);
1667     }
1668
1669     /**
1670      * The AutoFindPrimaryFailureResponseHandler automatically processes Failure responses when finding a primary
1671      * replica and sends a wrapped Failure response to some targetActor
1672      */
1673     private static abstract class AutoFindPrimaryFailureResponseHandler implements FindPrimaryResponseHandler {
1674         private final ActorRef targetActor;
1675         private final String shardName;
1676         private final String persistenceId;
1677         private final ActorRef shardManagerActor;
1678
1679         /**
1680          * @param targetActor The actor to whom the Failure response should be sent when a FindPrimary failure occurs
1681          * @param shardName The name of the shard for which the primary replica had to be found
1682          * @param persistenceId The persistenceId for the ShardManager
1683          * @param shardManagerActor The ShardManager actor which triggered the call to FindPrimary
1684          */
1685         protected AutoFindPrimaryFailureResponseHandler(ActorRef targetActor, String shardName, String persistenceId, ActorRef shardManagerActor){
1686             this.targetActor = Preconditions.checkNotNull(targetActor);
1687             this.shardName = Preconditions.checkNotNull(shardName);
1688             this.persistenceId = Preconditions.checkNotNull(persistenceId);
1689             this.shardManagerActor = Preconditions.checkNotNull(shardManagerActor);
1690         }
1691
1692         public ActorRef getTargetActor() {
1693             return targetActor;
1694         }
1695
1696         public String getShardName() {
1697             return shardName;
1698         }
1699
1700         @Override
1701         public void onFailure(Throwable failure) {
1702             LOG.debug ("{}: Received failure from FindPrimary for shard {}", persistenceId, shardName, failure);
1703             targetActor.tell(new Status.Failure(new RuntimeException(
1704                     String.format("Failed to find leader for shard %s", shardName), failure)), shardManagerActor);
1705         }
1706
1707         @Override
1708         public void onUnknownResponse(Object response) {
1709             String msg = String.format("Failed to find leader for shard %s: received response: %s",
1710                     shardName, response);
1711             LOG.debug ("{}: {}", persistenceId, msg);
1712             targetActor.tell(new Status.Failure(response instanceof Throwable ? (Throwable) response :
1713                     new RuntimeException(msg)), shardManagerActor);
1714         }
1715     }
1716
1717
1718     /**
1719      * The PrimaryShardFoundForContext is a DTO which puts together a message (aka 'Context' message) which needs to be
1720      * forwarded to the primary replica of a shard and the message (aka 'PrimaryShardFound' message) that is received
1721      * as a successful response to find primary.
1722      */
1723     private static class PrimaryShardFoundForContext {
1724         private final String shardName;
1725         private final Object contextMessage;
1726         private final RemotePrimaryShardFound remotePrimaryShardFound;
1727         private final LocalPrimaryShardFound localPrimaryShardFound;
1728
1729         public PrimaryShardFoundForContext(@Nonnull String shardName, @Nonnull Object contextMessage,
1730                 @Nonnull Object primaryFoundMessage) {
1731             this.shardName = Preconditions.checkNotNull(shardName);
1732             this.contextMessage = Preconditions.checkNotNull(contextMessage);
1733             Preconditions.checkNotNull(primaryFoundMessage);
1734             this.remotePrimaryShardFound = (primaryFoundMessage instanceof RemotePrimaryShardFound) ?
1735                     (RemotePrimaryShardFound) primaryFoundMessage : null;
1736             this.localPrimaryShardFound = (primaryFoundMessage instanceof LocalPrimaryShardFound) ?
1737                     (LocalPrimaryShardFound) primaryFoundMessage : null;
1738         }
1739
1740         @Nonnull
1741         String getPrimaryPath(){
1742             if(remotePrimaryShardFound != null) {
1743                 return remotePrimaryShardFound.getPrimaryPath();
1744             }
1745             return localPrimaryShardFound.getPrimaryPath();
1746         }
1747
1748         @Nonnull
1749         Object getContextMessage() {
1750             return contextMessage;
1751         }
1752
1753         @Nullable
1754         RemotePrimaryShardFound getRemotePrimaryShardFound() {
1755             return remotePrimaryShardFound;
1756         }
1757
1758         @Nonnull
1759         String getShardName() {
1760             return shardName;
1761         }
1762     }
1763
1764     /**
1765      * The WrappedShardResponse class wraps a response from a Shard.
1766      */
1767     private static class WrappedShardResponse {
1768         private final ShardIdentifier shardId;
1769         private final Object response;
1770         private final String leaderPath;
1771
1772         private WrappedShardResponse(ShardIdentifier shardId, Object response, String leaderPath) {
1773             this.shardId = shardId;
1774             this.response = response;
1775             this.leaderPath = leaderPath;
1776         }
1777
1778         ShardIdentifier getShardId() {
1779             return shardId;
1780         }
1781
1782         Object getResponse() {
1783             return response;
1784         }
1785
1786         String getLeaderPath() {
1787             return leaderPath;
1788         }
1789     }
1790 }
1791
1792
1793