opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.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;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.ActorSelection;
  13 import akka.actor.Cancellable;
  14 import akka.actor.Props;
  15 import akka.actor.Status.Failure;
  16 import akka.serialization.Serialization;
  17 import com.google.common.annotations.VisibleForTesting;
  18 import com.google.common.base.Optional;
  19 import com.google.common.base.Preconditions;
  20 import com.google.common.base.Ticker;
  21 import com.google.common.collect.ImmutableList;
  22 import com.google.common.collect.Range;
  23 import java.io.IOException;
  24 import java.util.Arrays;
  25 import java.util.Collection;
  26 import java.util.Collections;
  27 import java.util.HashMap;
  28 import java.util.Map;
  29 import java.util.concurrent.TimeUnit;
  30 import javax.annotation.Nonnull;
  31 import javax.annotation.Nullable;
  32 import org.opendaylight.controller.cluster.access.ABIVersion;
  33 import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
  34 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
  35 import org.opendaylight.controller.cluster.access.commands.LocalHistoryRequest;
  36 import org.opendaylight.controller.cluster.access.commands.NotLeaderException;
  37 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
  38 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
  39 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
  40 import org.opendaylight.controller.cluster.access.concepts.Request;
  41 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
  42 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  43 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
  44 import org.opendaylight.controller.cluster.access.concepts.RetiredGenerationException;
  45 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
  46 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  47 import org.opendaylight.controller.cluster.access.concepts.UnsupportedRequestException;
  48 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
  49 import org.opendaylight.controller.cluster.common.actor.MessageTracker;
  50 import org.opendaylight.controller.cluster.common.actor.MessageTracker.Error;
  51 import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
  52 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  53 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  54 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
  55 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
  56 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
  57 import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
  58 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  59 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  60 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
  61 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  62 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  63 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
  64 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
  65 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot.ShardSnapshot;
  66 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
  67 import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree;
  68 import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState;
  69 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
  70 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
  71 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
  72 import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
  73 import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
  74 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  75 import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
  76 import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
  77 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
  78 import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier;
  79 import org.opendaylight.controller.cluster.raft.RaftActor;
  80 import org.opendaylight.controller.cluster.raft.RaftActorRecoveryCohort;
  81 import org.opendaylight.controller.cluster.raft.RaftActorSnapshotCohort;
  82 import org.opendaylight.controller.cluster.raft.RaftState;
  83 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
  84 import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
  85 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  86 import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
  87 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  88 import org.opendaylight.yangtools.concepts.Identifier;
  89 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
  90 import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree;
  91 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
  92 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  93 import scala.concurrent.duration.Duration;
  94 import scala.concurrent.duration.FiniteDuration;
  95
  96 /**
  97  * A Shard represents a portion of the logical data tree.
  98  *
  99  * <p>
 100  * Our Shard uses InMemoryDataTree as it's internal representation and delegates all requests it
 101  */
 102 public class Shard extends RaftActor {
 103
 104     @VisibleForTesting
 105     static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = new Object() {
 106         @Override
 107         public String toString() {
 108             return "txCommitTimeoutCheck";
 109         }
 110     };
 111
 112     @VisibleForTesting
 113     static final Object GET_SHARD_MBEAN_MESSAGE = new Object() {
 114         @Override
 115         public String toString() {
 116             return "getShardMBeanMessage";
 117         }
 118     };
 119
 120     // FIXME: shard names should be encapsulated in their own class and this should be exposed as a constant.
 121     public static final String DEFAULT_NAME = "default";
 122
 123     private static final Collection<ABIVersion> SUPPORTED_ABIVERSIONS;
 124
 125     static {
 126         final ABIVersion[] values = ABIVersion.values();
 127         final ABIVersion[] real = Arrays.copyOfRange(values, 1, values.length - 1);
 128         SUPPORTED_ABIVERSIONS = ImmutableList.copyOf(real).reverse();
 129     }
 130
 131     // FIXME: make this a dynamic property based on mailbox size and maximum number of clients
 132     private static final int CLIENT_MAX_MESSAGES = 1000;
 133
 134     // The state of this Shard
 135     private final ShardDataTree store;
 136
 137     /// The name of this shard
 138     private final String name;
 139
 140     private final ShardStats shardMBean;
 141
 142     private DatastoreContext datastoreContext;
 143
 144     private final ShardCommitCoordinator commitCoordinator;
 145
 146     private long transactionCommitTimeout;
 147
 148     private Cancellable txCommitTimeoutCheckSchedule;
 149
 150     private final Optional<ActorRef> roleChangeNotifier;
 151
 152     private final MessageTracker appendEntriesReplyTracker;
 153
 154     private final ShardTransactionActorFactory transactionActorFactory;
 155
 156     private final ShardSnapshotCohort snapshotCohort;
 157
 158     private final DataTreeChangeListenerSupport treeChangeSupport = new DataTreeChangeListenerSupport(this);
 159     private final DataChangeListenerSupport changeSupport = new DataChangeListenerSupport(this);
 160
 161
 162     private ShardSnapshot restoreFromSnapshot;
 163
 164     private final ShardTransactionMessageRetrySupport messageRetrySupport;
 165
 166     private final FrontendMetadata frontendMetadata = new FrontendMetadata();
 167     private final Map<FrontendIdentifier, LeaderFrontendState> knownFrontends = new HashMap<>();
 168
 169     protected Shard(final AbstractBuilder<?, ?> builder) {
 170         super(builder.getId().toString(), builder.getPeerAddresses(),
 171                 Optional.of(builder.getDatastoreContext().getShardRaftConfig()), DataStoreVersions.CURRENT_VERSION);
 172
 173         this.name = builder.getId().toString();
 174         this.datastoreContext = builder.getDatastoreContext();
 175         this.restoreFromSnapshot = builder.getRestoreFromSnapshot();
 176
 177         setPersistence(datastoreContext.isPersistent());
 178
 179         LOG.info("Shard created : {}, persistent : {}", name, datastoreContext.isPersistent());
 180
 181         ShardDataTreeChangeListenerPublisherActorProxy treeChangeListenerPublisher =
 182                 new ShardDataTreeChangeListenerPublisherActorProxy(getContext(), name + "-DTCL-publisher");
 183         ShardDataChangeListenerPublisherActorProxy dataChangeListenerPublisher =
 184                 new ShardDataChangeListenerPublisherActorProxy(getContext(), name + "-DCL-publisher");
 185         if (builder.getDataTree() != null) {
 186             store = new ShardDataTree(this, builder.getSchemaContext(), builder.getDataTree(),
 187                     treeChangeListenerPublisher, dataChangeListenerPublisher, name);
 188         } else {
 189             store = new ShardDataTree(this, builder.getSchemaContext(), builder.getTreeType(),
 190                     builder.getDatastoreContext().getStoreRoot(), treeChangeListenerPublisher,
 191                     dataChangeListenerPublisher, name);
 192         }
 193
 194         shardMBean = ShardMBeanFactory.getShardStatsMBean(name, datastoreContext.getDataStoreMXBeanType(), this);
 195
 196         if (isMetricsCaptureEnabled()) {
 197             getContext().become(new MeteringBehavior(this));
 198         }
 199
 200         commitCoordinator = new ShardCommitCoordinator(store, LOG, this.name);
 201
 202         setTransactionCommitTimeout();
 203
 204         // create a notifier actor for each cluster member
 205         roleChangeNotifier = createRoleChangeNotifier(name);
 206
 207         appendEntriesReplyTracker = new MessageTracker(AppendEntriesReply.class,
 208                 getRaftActorContext().getConfigParams().getIsolatedCheckIntervalInMillis());
 209
 210         transactionActorFactory = new ShardTransactionActorFactory(store, datastoreContext,
 211                 new Dispatchers(context().system().dispatchers()).getDispatcherPath(
 212                         Dispatchers.DispatcherType.Transaction), self(), getContext(), shardMBean);
 213
 214         snapshotCohort = ShardSnapshotCohort.create(getContext(), builder.getId().getMemberName(), store, LOG,
 215             this.name);
 216
 217         messageRetrySupport = new ShardTransactionMessageRetrySupport(this);
 218     }
 219
 220     private void setTransactionCommitTimeout() {
 221         transactionCommitTimeout = TimeUnit.MILLISECONDS.convert(
 222                 datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS) / 2;
 223     }
 224
 225     private Optional<ActorRef> createRoleChangeNotifier(final String shardId) {
 226         ActorRef shardRoleChangeNotifier = this.getContext().actorOf(
 227             RoleChangeNotifier.getProps(shardId), shardId + "-notifier");
 228         return Optional.of(shardRoleChangeNotifier);
 229     }
 230
 231     @Override
 232     public void postStop() {
 233         LOG.info("Stopping Shard {}", persistenceId());
 234
 235         super.postStop();
 236
 237         messageRetrySupport.close();
 238
 239         if (txCommitTimeoutCheckSchedule != null) {
 240             txCommitTimeoutCheckSchedule.cancel();
 241         }
 242
 243         commitCoordinator.abortPendingTransactions("Transaction aborted due to shutdown.", this);
 244
 245         shardMBean.unregisterMBean();
 246     }
 247
 248     @Override
 249     protected void handleRecover(final Object message) {
 250         LOG.debug("{}: onReceiveRecover: Received message {} from {}", persistenceId(), message.getClass(),
 251             getSender());
 252
 253         super.handleRecover(message);
 254         if (LOG.isTraceEnabled()) {
 255             appendEntriesReplyTracker.begin();
 256         }
 257     }
 258
 259     @SuppressWarnings("checkstyle:IllegalCatch")
 260     @Override
 261     protected void handleNonRaftCommand(final Object message) {
 262         try (final MessageTracker.Context context = appendEntriesReplyTracker.received(message)) {
 263             final Optional<Error> maybeError = context.error();
 264             if (maybeError.isPresent()) {
 265                 LOG.trace("{} : AppendEntriesReply failed to arrive at the expected interval {}", persistenceId(),
 266                     maybeError.get());
 267             }
 268
 269             if (message instanceof RequestEnvelope) {
 270                 final long now = ticker().read();
 271                 final RequestEnvelope envelope = (RequestEnvelope)message;
 272
 273                 try {
 274                     final RequestSuccess<?, ?> success = handleRequest(envelope, now);
 275                     if (success != null) {
 276                         envelope.sendSuccess(success, ticker().read() - now);
 277                     }
 278                 } catch (RequestException e) {
 279                     LOG.debug("{}: request {} failed", persistenceId(), envelope, e);
 280                     envelope.sendFailure(e, ticker().read() - now);
 281                 } catch (Exception e) {
 282                     LOG.debug("{}: request {} caused failure", persistenceId(), envelope, e);
 283                     envelope.sendFailure(new RuntimeRequestException("Request failed to process", e),
 284                         ticker().read() - now);
 285                 }
 286             } else if (message instanceof ConnectClientRequest) {
 287                 handleConnectClient((ConnectClientRequest)message);
 288             } else if (CreateTransaction.isSerializedType(message)) {
 289                 handleCreateTransaction(message);
 290             } else if (message instanceof BatchedModifications) {
 291                 handleBatchedModifications((BatchedModifications)message);
 292             } else if (message instanceof ForwardedReadyTransaction) {
 293                 handleForwardedReadyTransaction((ForwardedReadyTransaction) message);
 294             } else if (message instanceof ReadyLocalTransaction) {
 295                 handleReadyLocalTransaction((ReadyLocalTransaction)message);
 296             } else if (CanCommitTransaction.isSerializedType(message)) {
 297                 handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message));
 298             } else if (CommitTransaction.isSerializedType(message)) {
 299                 handleCommitTransaction(CommitTransaction.fromSerializable(message));
 300             } else if (AbortTransaction.isSerializedType(message)) {
 301                 handleAbortTransaction(AbortTransaction.fromSerializable(message));
 302             } else if (CloseTransactionChain.isSerializedType(message)) {
 303                 closeTransactionChain(CloseTransactionChain.fromSerializable(message));
 304             } else if (message instanceof RegisterChangeListener) {
 305                 changeSupport.onMessage((RegisterChangeListener) message, isLeader(), hasLeader());
 306             } else if (message instanceof RegisterDataTreeChangeListener) {
 307                 treeChangeSupport.onMessage((RegisterDataTreeChangeListener) message, isLeader(), hasLeader());
 308             } else if (message instanceof UpdateSchemaContext) {
 309                 updateSchemaContext((UpdateSchemaContext) message);
 310             } else if (message instanceof PeerAddressResolved) {
 311                 PeerAddressResolved resolved = (PeerAddressResolved) message;
 312                 setPeerAddress(resolved.getPeerId(), resolved.getPeerAddress());
 313             } else if (TX_COMMIT_TIMEOUT_CHECK_MESSAGE.equals(message)) {
 314                 store.checkForExpiredTransactions(transactionCommitTimeout);
 315                 commitCoordinator.checkForExpiredTransactions(transactionCommitTimeout, this);
 316             } else if (message instanceof DatastoreContext) {
 317                 onDatastoreContext((DatastoreContext)message);
 318             } else if (message instanceof RegisterRoleChangeListener) {
 319                 roleChangeNotifier.get().forward(message, context());
 320             } else if (message instanceof FollowerInitialSyncUpStatus) {
 321                 shardMBean.setFollowerInitialSyncStatus(((FollowerInitialSyncUpStatus) message).isInitialSyncDone());
 322                 context().parent().tell(message, self());
 323             } else if (GET_SHARD_MBEAN_MESSAGE.equals(message)) {
 324                 sender().tell(getShardMBean(), self());
 325             } else if (message instanceof GetShardDataTree) {
 326                 sender().tell(store.getDataTree(), self());
 327             } else if (message instanceof ServerRemoved) {
 328                 context().parent().forward(message, context());
 329             } else if (ShardTransactionMessageRetrySupport.TIMER_MESSAGE_CLASS.isInstance(message)) {
 330                 messageRetrySupport.onTimerMessage(message);
 331             } else if (message instanceof DataTreeCohortActorRegistry.CohortRegistryCommand) {
 332                 store.processCohortRegistryCommand(getSender(),
 333                         (DataTreeCohortActorRegistry.CohortRegistryCommand) message);
 334             } else {
 335                 super.handleNonRaftCommand(message);
 336             }
 337         }
 338     }
 339
 340     // Acquire our frontend tracking handle and verify generation matches
 341     private LeaderFrontendState getFrontend(final ClientIdentifier clientId) throws RequestException {
 342         final LeaderFrontendState existing = knownFrontends.get(clientId.getFrontendId());
 343         if (existing != null) {
 344             final int cmp = Long.compareUnsigned(existing.getIdentifier().getGeneration(), clientId.getGeneration());
 345             if (cmp == 0) {
 346                 return existing;
 347             }
 348             if (cmp > 0) {
 349                 LOG.debug("{}: rejecting request from outdated client {}", persistenceId(), clientId);
 350                 throw new RetiredGenerationException(existing.getIdentifier().getGeneration());
 351             }
 352
 353             LOG.info("{}: retiring state {}, outdated by request from client {}", persistenceId(), existing, clientId);
 354             existing.retire();
 355             knownFrontends.remove(clientId.getFrontendId());
 356         } else {
 357             LOG.debug("{}: client {} is not yet known", persistenceId(), clientId);
 358         }
 359
 360         final LeaderFrontendState ret = new LeaderFrontendState(persistenceId(), clientId, store);
 361         knownFrontends.put(clientId.getFrontendId(), ret);
 362         LOG.debug("{}: created state {} for client {}", persistenceId(), ret, clientId);
 363         return ret;
 364     }
 365
 366     private static @Nonnull ABIVersion selectVersion(final ConnectClientRequest message) {
 367         final Range<ABIVersion> clientRange = Range.closed(message.getMinVersion(), message.getMaxVersion());
 368         for (ABIVersion v : SUPPORTED_ABIVERSIONS) {
 369             if (clientRange.contains(v)) {
 370                 return v;
 371             }
 372         }
 373
 374         throw new IllegalArgumentException(String.format(
 375             "No common version between backend versions %s and client versions %s", SUPPORTED_ABIVERSIONS,
 376             clientRange));
 377     }
 378
 379     @SuppressWarnings("checkstyle:IllegalCatch")
 380     private void handleConnectClient(final ConnectClientRequest message) {
 381         try {
 382             if (!isLeader() || !isLeaderActive()) {
 383                 LOG.debug("{}: not currently leader, rejecting request {}", persistenceId(), message);
 384                 throw new NotLeaderException(getSelf());
 385             }
 386
 387             final ABIVersion selectedVersion = selectVersion(message);
 388             final LeaderFrontendState frontend = getFrontend(message.getTarget());
 389             frontend.reconnect();
 390             message.getReplyTo().tell(new ConnectClientSuccess(message.getTarget(), message.getSequence(), getSelf(),
 391                 ImmutableList.of(), store.getDataTree(), CLIENT_MAX_MESSAGES).toVersion(selectedVersion),
 392                 ActorRef.noSender());
 393         } catch (RequestException | RuntimeException e) {
 394             message.getReplyTo().tell(new Failure(e), ActorRef.noSender());
 395         }
 396     }
 397
 398     private @Nullable RequestSuccess<?, ?> handleRequest(final RequestEnvelope envelope, final long now)
 399             throws RequestException {
 400         // We are not the leader, hence we want to fail-fast.
 401         if (!isLeader() || !isLeaderActive()) {
 402             LOG.debug("{}: not currently leader, rejecting request {}", persistenceId(), envelope);
 403             throw new NotLeaderException(getSelf());
 404         }
 405
 406         final Request<?, ?> request = envelope.getMessage();
 407         if (request instanceof TransactionRequest) {
 408             final TransactionRequest<?> txReq = (TransactionRequest<?>)request;
 409             final ClientIdentifier clientId = txReq.getTarget().getHistoryId().getClientId();
 410             return getFrontend(clientId).handleTransactionRequest(txReq, envelope, now);
 411         } else if (request instanceof LocalHistoryRequest) {
 412             final LocalHistoryRequest<?> lhReq = (LocalHistoryRequest<?>)request;
 413             final ClientIdentifier clientId = lhReq.getTarget().getClientId();
 414             return getFrontend(clientId).handleLocalHistoryRequest(lhReq, envelope, now);
 415         } else {
 416             LOG.debug("{}: rejecting unsupported request {}", persistenceId(), request);
 417             throw new UnsupportedRequestException(request);
 418         }
 419     }
 420
 421     private boolean hasLeader() {
 422         return getLeaderId() != null;
 423     }
 424
 425     public int getPendingTxCommitQueueSize() {
 426         return store.getQueueSize();
 427     }
 428
 429     public int getCohortCacheSize() {
 430         return commitCoordinator.getCohortCacheSize();
 431     }
 432
 433     @Override
 434     protected Optional<ActorRef> getRoleChangeNotifier() {
 435         return roleChangeNotifier;
 436     }
 437
 438     @Override
 439     protected LeaderStateChanged newLeaderStateChanged(final String memberId, final String leaderId,
 440             final short leaderPayloadVersion) {
 441         return isLeader() ? new ShardLeaderStateChanged(memberId, leaderId, store.getDataTree(), leaderPayloadVersion)
 442                 : new ShardLeaderStateChanged(memberId, leaderId, leaderPayloadVersion);
 443     }
 444
 445     protected void onDatastoreContext(final DatastoreContext context) {
 446         datastoreContext = context;
 447
 448         setTransactionCommitTimeout();
 449
 450         setPersistence(datastoreContext.isPersistent());
 451
 452         updateConfigParams(datastoreContext.getShardRaftConfig());
 453     }
 454
 455     // applyState() will be invoked once consensus is reached on the payload
 456     void persistPayload(final TransactionIdentifier transactionId, final Payload payload, boolean batchHint) {
 457         boolean canSkipPayload = !hasFollowers() && !persistence().isRecoveryApplicable();
 458         if (canSkipPayload) {
 459             applyState(self(), transactionId, payload);
 460         } else {
 461             // We are faking the sender
 462             persistData(self(), transactionId, payload, batchHint);
 463         }
 464     }
 465
 466     private void handleCommitTransaction(final CommitTransaction commit) {
 467         if (isLeader()) {
 468             commitCoordinator.handleCommit(commit.getTransactionId(), getSender(), this);
 469         } else {
 470             ActorSelection leader = getLeader();
 471             if (leader == null) {
 472                 messageRetrySupport.addMessageToRetry(commit, getSender(),
 473                         "Could not commit transaction " + commit.getTransactionId());
 474             } else {
 475                 LOG.debug("{}: Forwarding CommitTransaction to leader {}", persistenceId(), leader);
 476                 leader.forward(commit, getContext());
 477             }
 478         }
 479     }
 480
 481     private void handleCanCommitTransaction(final CanCommitTransaction canCommit) {
 482         LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionId());
 483
 484         if (isLeader()) {
 485             commitCoordinator.handleCanCommit(canCommit.getTransactionId(), getSender(), this);
 486         } else {
 487             ActorSelection leader = getLeader();
 488             if (leader == null) {
 489                 messageRetrySupport.addMessageToRetry(canCommit, getSender(),
 490                         "Could not canCommit transaction " + canCommit.getTransactionId());
 491             } else {
 492                 LOG.debug("{}: Forwarding CanCommitTransaction to leader {}", persistenceId(), leader);
 493                 leader.forward(canCommit, getContext());
 494             }
 495         }
 496     }
 497
 498     @SuppressWarnings("checkstyle:IllegalCatch")
 499     protected void handleBatchedModificationsLocal(final BatchedModifications batched, final ActorRef sender) {
 500         try {
 501             commitCoordinator.handleBatchedModifications(batched, sender, this);
 502         } catch (Exception e) {
 503             LOG.error("{}: Error handling BatchedModifications for Tx {}", persistenceId(),
 504                     batched.getTransactionId(), e);
 505             sender.tell(new Failure(e), getSelf());
 506         }
 507     }
 508
 509     private void handleBatchedModifications(final BatchedModifications batched) {
 510         // This message is sent to prepare the modifications transaction directly on the Shard as an
 511         // optimization to avoid the extra overhead of a separate ShardTransaction actor. On the last
 512         // BatchedModifications message, the caller sets the ready flag in the message indicating
 513         // modifications are complete. The reply contains the cohort actor path (this actor) for the caller
 514         // to initiate the 3-phase commit. This also avoids the overhead of sending an additional
 515         // ReadyTransaction message.
 516
 517         // If we're not the leader then forward to the leader. This is a safety measure - we shouldn't
 518         // normally get here if we're not the leader as the front-end (TransactionProxy) should determine
 519         // the primary/leader shard. However with timing and caching on the front-end, there's a small
 520         // window where it could have a stale leader during leadership transitions.
 521         //
 522         boolean isLeaderActive = isLeaderActive();
 523         if (isLeader() && isLeaderActive) {
 524             handleBatchedModificationsLocal(batched, getSender());
 525         } else {
 526             ActorSelection leader = getLeader();
 527             if (!isLeaderActive || leader == null) {
 528                 messageRetrySupport.addMessageToRetry(batched, getSender(),
 529                         "Could not commit transaction " + batched.getTransactionId());
 530             } else {
 531                 // If this is not the first batch and leadership changed in between batched messages,
 532                 // we need to reconstruct previous BatchedModifications from the transaction
 533                 // DataTreeModification, honoring the max batched modification count, and forward all the
 534                 // previous BatchedModifications to the new leader.
 535                 Collection<BatchedModifications> newModifications = commitCoordinator
 536                         .createForwardedBatchedModifications(batched,
 537                                 datastoreContext.getShardBatchedModificationCount());
 538
 539                 LOG.debug("{}: Forwarding {} BatchedModifications to leader {}", persistenceId(),
 540                         newModifications.size(), leader);
 541
 542                 for (BatchedModifications bm : newModifications) {
 543                     leader.forward(bm, getContext());
 544                 }
 545             }
 546         }
 547     }
 548
 549     private boolean failIfIsolatedLeader(final ActorRef sender) {
 550         if (isIsolatedLeader()) {
 551             sender.tell(new Failure(new NoShardLeaderException(String.format(
 552                     "Shard %s was the leader but has lost contact with all of its followers. Either all"
 553                     + " other follower nodes are down or this node is isolated by a network partition.",
 554                     persistenceId()))), getSelf());
 555             return true;
 556         }
 557
 558         return false;
 559     }
 560
 561     protected boolean isIsolatedLeader() {
 562         return getRaftState() == RaftState.IsolatedLeader;
 563     }
 564
 565     @SuppressWarnings("checkstyle:IllegalCatch")
 566     private void handleReadyLocalTransaction(final ReadyLocalTransaction message) {
 567         LOG.debug("{}: handleReadyLocalTransaction for {}", persistenceId(), message.getTransactionId());
 568
 569         boolean isLeaderActive = isLeaderActive();
 570         if (isLeader() && isLeaderActive) {
 571             try {
 572                 commitCoordinator.handleReadyLocalTransaction(message, getSender(), this);
 573             } catch (Exception e) {
 574                 LOG.error("{}: Error handling ReadyLocalTransaction for Tx {}", persistenceId(),
 575                         message.getTransactionId(), e);
 576                 getSender().tell(new Failure(e), getSelf());
 577             }
 578         } else {
 579             ActorSelection leader = getLeader();
 580             if (!isLeaderActive || leader == null) {
 581                 messageRetrySupport.addMessageToRetry(message, getSender(),
 582                         "Could not commit transaction " + message.getTransactionId());
 583             } else {
 584                 LOG.debug("{}: Forwarding ReadyLocalTransaction to leader {}", persistenceId(), leader);
 585                 message.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion());
 586                 leader.forward(message, getContext());
 587             }
 588         }
 589     }
 590
 591     private void handleForwardedReadyTransaction(final ForwardedReadyTransaction forwardedReady) {
 592         LOG.debug("{}: handleForwardedReadyTransaction for {}", persistenceId(), forwardedReady.getTransactionId());
 593
 594         boolean isLeaderActive = isLeaderActive();
 595         if (isLeader() && isLeaderActive) {
 596             commitCoordinator.handleForwardedReadyTransaction(forwardedReady, getSender(), this);
 597         } else {
 598             ActorSelection leader = getLeader();
 599             if (!isLeaderActive || leader == null) {
 600                 messageRetrySupport.addMessageToRetry(forwardedReady, getSender(),
 601                         "Could not commit transaction " + forwardedReady.getTransactionId());
 602             } else {
 603                 LOG.debug("{}: Forwarding ForwardedReadyTransaction to leader {}", persistenceId(), leader);
 604
 605                 ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(forwardedReady.getTransactionId(),
 606                         forwardedReady.getTransaction().getSnapshot(), forwardedReady.isDoImmediateCommit());
 607                 readyLocal.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion());
 608                 leader.forward(readyLocal, getContext());
 609             }
 610         }
 611     }
 612
 613     private void handleAbortTransaction(final AbortTransaction abort) {
 614         doAbortTransaction(abort.getTransactionId(), getSender());
 615     }
 616
 617     void doAbortTransaction(final TransactionIdentifier transactionID, final ActorRef sender) {
 618         commitCoordinator.handleAbort(transactionID, sender, this);
 619     }
 620
 621     private void handleCreateTransaction(final Object message) {
 622         if (isLeader()) {
 623             createTransaction(CreateTransaction.fromSerializable(message));
 624         } else if (getLeader() != null) {
 625             getLeader().forward(message, getContext());
 626         } else {
 627             getSender().tell(new Failure(new NoShardLeaderException(
 628                     "Could not create a shard transaction", persistenceId())), getSelf());
 629         }
 630     }
 631
 632     private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) {
 633         store.closeTransactionChain(closeTransactionChain.getIdentifier());
 634     }
 635
 636     @SuppressWarnings("checkstyle:IllegalCatch")
 637     private void createTransaction(final CreateTransaction createTransaction) {
 638         try {
 639             if (TransactionType.fromInt(createTransaction.getTransactionType()) != TransactionType.READ_ONLY
 640                     && failIfIsolatedLeader(getSender())) {
 641                 return;
 642             }
 643
 644             ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(),
 645                 createTransaction.getTransactionId());
 646
 647             getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor),
 648                     createTransaction.getTransactionId(), createTransaction.getVersion()).toSerializable(), getSelf());
 649         } catch (Exception e) {
 650             getSender().tell(new Failure(e), getSelf());
 651         }
 652     }
 653
 654     private ActorRef createTransaction(final int transactionType, final TransactionIdentifier transactionId) {
 655         LOG.debug("{}: Creating transaction : {} ", persistenceId(), transactionId);
 656         return transactionActorFactory.newShardTransaction(TransactionType.fromInt(transactionType),
 657             transactionId);
 658     }
 659
 660     private void updateSchemaContext(final UpdateSchemaContext message) {
 661         updateSchemaContext(message.getSchemaContext());
 662     }
 663
 664     @VisibleForTesting
 665     void updateSchemaContext(final SchemaContext schemaContext) {
 666         store.updateSchemaContext(schemaContext);
 667     }
 668
 669     private boolean isMetricsCaptureEnabled() {
 670         CommonConfig config = new CommonConfig(getContext().system().settings().config());
 671         return config.isMetricCaptureEnabled();
 672     }
 673
 674     @Override
 675     @VisibleForTesting
 676     public RaftActorSnapshotCohort getRaftActorSnapshotCohort() {
 677         return snapshotCohort;
 678     }
 679
 680     @Override
 681     @Nonnull
 682     protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() {
 683         return new ShardRecoveryCoordinator(store,
 684             restoreFromSnapshot != null ? restoreFromSnapshot.getSnapshot() : null, persistenceId(), LOG);
 685     }
 686
 687     @Override
 688     protected void onRecoveryComplete() {
 689         restoreFromSnapshot = null;
 690
 691         //notify shard manager
 692         getContext().parent().tell(new ActorInitialized(), getSelf());
 693
 694         // Being paranoid here - this method should only be called once but just in case...
 695         if (txCommitTimeoutCheckSchedule == null) {
 696             // Schedule a message to be periodically sent to check if the current in-progress
 697             // transaction should be expired and aborted.
 698             FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS);
 699             txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule(
 700                     period, period, getSelf(),
 701                     TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender());
 702         }
 703     }
 704
 705     @Override
 706     protected void applyState(final ActorRef clientActor, final Identifier identifier, final Object data) {
 707         if (data instanceof Payload) {
 708             try {
 709                 store.applyReplicatedPayload(identifier, (Payload)data);
 710             } catch (DataValidationFailedException | IOException e) {
 711                 LOG.error("{}: Error applying replica {}", persistenceId(), identifier, e);
 712             }
 713         } else {
 714             LOG.error("{}: Unknown state for {} received {}", persistenceId(), identifier, data);
 715         }
 716     }
 717
 718     @Override
 719     protected void onStateChanged() {
 720         boolean isLeader = isLeader();
 721         boolean hasLeader = hasLeader();
 722         changeSupport.onLeadershipChange(isLeader, hasLeader);
 723         treeChangeSupport.onLeadershipChange(isLeader, hasLeader);
 724
 725         // If this actor is no longer the leader close all the transaction chains
 726         if (!isLeader) {
 727             if (LOG.isDebugEnabled()) {
 728                 LOG.debug(
 729                     "{}: onStateChanged: Closing all transaction chains because shard {} is no longer the leader",
 730                     persistenceId(), getId());
 731             }
 732
 733             store.closeAllTransactionChains();
 734         }
 735
 736         if (hasLeader && !isIsolatedLeader()) {
 737             messageRetrySupport.retryMessages();
 738         }
 739     }
 740
 741     @Override
 742     protected void onLeaderChanged(final String oldLeader, final String newLeader) {
 743         shardMBean.incrementLeadershipChangeCount();
 744
 745         boolean hasLeader = hasLeader();
 746         if (hasLeader && !isLeader()) {
 747             // Another leader was elected. If we were the previous leader and had pending transactions, convert
 748             // them to transaction messages and send to the new leader.
 749             ActorSelection leader = getLeader();
 750             if (leader != null) {
 751                 Collection<?> messagesToForward = convertPendingTransactionsToMessages();
 752
 753                 if (!messagesToForward.isEmpty()) {
 754                     LOG.debug("{}: Forwarding {} pending transaction messages to leader {}", persistenceId(),
 755                             messagesToForward.size(), leader);
 756
 757                     for (Object message : messagesToForward) {
 758                         leader.tell(message, self());
 759                     }
 760                 }
 761             } else {
 762                 commitCoordinator.abortPendingTransactions("The transacton was aborted due to inflight leadership "
 763                         + "change and the leader address isn't available.", this);
 764             }
 765         }
 766
 767         if (hasLeader && !isIsolatedLeader()) {
 768             messageRetrySupport.retryMessages();
 769         }
 770     }
 771
 772     /**
 773      * Clears all pending transactions and converts them to messages to be forwarded to a new leader.
 774      *
 775      * @return the converted messages
 776      */
 777     public Collection<?> convertPendingTransactionsToMessages() {
 778         return commitCoordinator.convertPendingTransactionsToMessages(
 779                 datastoreContext.getShardBatchedModificationCount());
 780     }
 781
 782     @Override
 783     protected void pauseLeader(final Runnable operation) {
 784         LOG.debug("{}: In pauseLeader, operation: {}", persistenceId(), operation);
 785         store.setRunOnPendingTransactionsComplete(operation);
 786     }
 787
 788     @Override
 789     protected OnDemandRaftState.AbstractBuilder<?> newOnDemandRaftStateBuilder() {
 790         return OnDemandShardState.newBuilder().treeChangeListenerActors(treeChangeSupport.getListenerActors())
 791                 .dataChangeListenerActors(changeSupport.getListenerActors())
 792                 .commitCohortActors(store.getCohortActors());
 793     }
 794
 795     @Override
 796     public String persistenceId() {
 797         return this.name;
 798     }
 799
 800     @VisibleForTesting
 801     ShardCommitCoordinator getCommitCoordinator() {
 802         return commitCoordinator;
 803     }
 804
 805     public DatastoreContext getDatastoreContext() {
 806         return datastoreContext;
 807     }
 808
 809     @VisibleForTesting
 810     public ShardDataTree getDataStore() {
 811         return store;
 812     }
 813
 814     @VisibleForTesting
 815     ShardStats getShardMBean() {
 816         return shardMBean;
 817     }
 818
 819     public static Builder builder() {
 820         return new Builder();
 821     }
 822
 823     public abstract static class AbstractBuilder<T extends AbstractBuilder<T, S>, S extends Shard> {
 824         private final Class<S> shardClass;
 825         private ShardIdentifier id;
 826         private Map<String, String> peerAddresses = Collections.emptyMap();
 827         private DatastoreContext datastoreContext;
 828         private SchemaContext schemaContext;
 829         private DatastoreSnapshot.ShardSnapshot restoreFromSnapshot;
 830         private TipProducingDataTree dataTree;
 831         private volatile boolean sealed;
 832
 833         protected AbstractBuilder(final Class<S> shardClass) {
 834             this.shardClass = shardClass;
 835         }
 836
 837         protected void checkSealed() {
 838             Preconditions.checkState(!sealed, "Builder isalready sealed - further modifications are not allowed");
 839         }
 840
 841         @SuppressWarnings("unchecked")
 842         private T self() {
 843             return (T) this;
 844         }
 845
 846         public T id(final ShardIdentifier newId) {
 847             checkSealed();
 848             this.id = newId;
 849             return self();
 850         }
 851
 852         public T peerAddresses(final Map<String, String> newPeerAddresses) {
 853             checkSealed();
 854             this.peerAddresses = newPeerAddresses;
 855             return self();
 856         }
 857
 858         public T datastoreContext(final DatastoreContext newDatastoreContext) {
 859             checkSealed();
 860             this.datastoreContext = newDatastoreContext;
 861             return self();
 862         }
 863
 864         public T schemaContext(final SchemaContext newSchemaContext) {
 865             checkSealed();
 866             this.schemaContext = newSchemaContext;
 867             return self();
 868         }
 869
 870         public T restoreFromSnapshot(final DatastoreSnapshot.ShardSnapshot newRestoreFromSnapshot) {
 871             checkSealed();
 872             this.restoreFromSnapshot = newRestoreFromSnapshot;
 873             return self();
 874         }
 875
 876         public T dataTree(final TipProducingDataTree newDataTree) {
 877             checkSealed();
 878             this.dataTree = newDataTree;
 879             return self();
 880         }
 881
 882         public ShardIdentifier getId() {
 883             return id;
 884         }
 885
 886         public Map<String, String> getPeerAddresses() {
 887             return peerAddresses;
 888         }
 889
 890         public DatastoreContext getDatastoreContext() {
 891             return datastoreContext;
 892         }
 893
 894         public SchemaContext getSchemaContext() {
 895             return schemaContext;
 896         }
 897
 898         public DatastoreSnapshot.ShardSnapshot getRestoreFromSnapshot() {
 899             return restoreFromSnapshot;
 900         }
 901
 902         public TipProducingDataTree getDataTree() {
 903             return dataTree;
 904         }
 905
 906         public TreeType getTreeType() {
 907             switch (datastoreContext.getLogicalStoreType()) {
 908                 case CONFIGURATION:
 909                     return TreeType.CONFIGURATION;
 910                 case OPERATIONAL:
 911                     return TreeType.OPERATIONAL;
 912                 default:
 913                     throw new IllegalStateException("Unhandled logical store type "
 914                             + datastoreContext.getLogicalStoreType());
 915             }
 916         }
 917
 918         protected void verify() {
 919             Preconditions.checkNotNull(id, "id should not be null");
 920             Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
 921             Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null");
 922             Preconditions.checkNotNull(schemaContext, "schemaContext should not be null");
 923         }
 924
 925         public Props props() {
 926             sealed = true;
 927             verify();
 928             return Props.create(shardClass, this);
 929         }
 930     }
 931
 932     public static class Builder extends AbstractBuilder<Builder, Shard> {
 933         private Builder() {
 934             super(Shard.class);
 935         }
 936     }
 937
 938     Ticker ticker() {
 939         return Ticker.systemTicker();
 940     }
 941 }