X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=7073ea758b2b751d93708914bfc3935779076908;hb=925cb4a228d0fda99c7bfeb432eb25285a223887;hp=3934489646530c136d9cc3bfc2e41e8028cbb9ba;hpb=1b8f7c7beaed83797320686bebddd536637aed9a;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java index 3934489646..d52d3a3bf2 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java @@ -10,688 +10,791 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorRef; import akka.actor.ActorSelection; -import akka.actor.PoisonPill; +import akka.actor.Cancellable; import akka.actor.Props; -import akka.event.Logging; -import akka.event.LoggingAdapter; -import akka.japi.Creator; -import akka.persistence.RecoveryFailure; import akka.serialization.Serialization; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Optional; import com.google.common.base.Preconditions; -import com.google.common.collect.Lists; -import com.google.common.util.concurrent.CheckedFuture; -import com.google.common.util.concurrent.FutureCallback; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.protobuf.ByteString; -import com.google.protobuf.InvalidProtocolBufferException; +import com.google.common.base.Ticker; +import java.io.IOException; +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nonnull; +import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.common.actor.CommonConfig; +import org.opendaylight.controller.cluster.common.actor.MessageTracker; +import org.opendaylight.controller.cluster.common.actor.MessageTracker.Error; import org.opendaylight.controller.cluster.common.actor.MeteringBehavior; +import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier; -import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier; import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory; import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; +import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized; +import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; +import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain; -import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; +import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply; -import org.opendaylight.controller.cluster.datastore.messages.EnableNotification; -import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction; +import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot; +import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot.ShardSnapshot; +import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction; +import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree; import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved; -import org.opendaylight.controller.cluster.datastore.messages.ReadData; -import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply; +import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener; -import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply; +import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener; +import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; -import org.opendaylight.controller.cluster.datastore.modification.Modification; -import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification; -import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec; +import org.opendaylight.controller.cluster.datastore.utils.Dispatchers; +import org.opendaylight.controller.cluster.notifications.LeaderStateChanged; +import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener; +import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier; import org.opendaylight.controller.cluster.raft.RaftActor; -import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; -import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; -import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload; +import org.opendaylight.controller.cluster.raft.RaftActorRecoveryCohort; +import org.opendaylight.controller.cluster.raft.RaftActorSnapshotCohort; +import org.opendaylight.controller.cluster.raft.RaftState; +import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus; +import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; +import org.opendaylight.controller.cluster.raft.messages.ServerRemoved; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; -import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener; -import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; -import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore; -import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory; -import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; -import org.opendaylight.yangtools.concepts.ListenerRegistration; -import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; -import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; +import org.opendaylight.yangtools.concepts.Identifier; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException; +import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree; +import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType; import org.opendaylight.yangtools.yang.model.api.SchemaContext; -import java.util.ArrayList; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ExecutionException; +import scala.concurrent.duration.Duration; +import scala.concurrent.duration.FiniteDuration; /** - * A Shard represents a portion of the logical data tree
- *

- * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it - *

+ * A Shard represents a portion of the logical data tree. + *

+ * Our Shard uses InMemoryDataTree as it's internal representation and delegates all requests it */ public class Shard extends RaftActor { + @VisibleForTesting + static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = new Object() { + @Override + public String toString() { + return "txCommitTimeoutCheck"; + } + }; + + @VisibleForTesting + static final Object GET_SHARD_MBEAN_MESSAGE = new Object() { + @Override + public String toString() { + return "getShardMBeanMessage"; + } + }; + + // FIXME: shard names should be encapsulated in their own class and this should be exposed as a constant. public static final String DEFAULT_NAME = "default"; // The state of this Shard - private final InMemoryDOMDataStore store; + private final ShardDataTree store; - private final Map - modificationToCohort = new HashMap<>(); + /// The name of this shard + private final String name; - private final LoggingAdapter LOG = - Logging.getLogger(getContext().system(), this); + private final ShardStats shardMBean; - // By default persistent will be true and can be turned off using the system - // property shard.persistent - private final boolean persistent; + private DatastoreContext datastoreContext; - /// The name of this shard - private final ShardIdentifier name; + private final ShardCommitCoordinator commitCoordinator; - private final ShardStats shardMBean; + private long transactionCommitTimeout; - private final List dataChangeListeners = new ArrayList<>(); + private Cancellable txCommitTimeoutCheckSchedule; - private final DatastoreContext datastoreContext; + private final Optional roleChangeNotifier; - private SchemaContext schemaContext; + private final MessageTracker appendEntriesReplyTracker; - private ActorRef createSnapshotTransaction; + private final ShardTransactionActorFactory transactionActorFactory; - /** - * Coordinates persistence recovery on startup. - */ - private ShardRecoveryCoordinator recoveryCoordinator; - private List currentLogRecoveryBatch; + private final ShardSnapshotCohort snapshotCohort; - private final Map transactionChains = new HashMap<>(); + private final DataTreeChangeListenerSupport treeChangeSupport = new DataTreeChangeListenerSupport(this); + private final DataChangeListenerSupport changeSupport = new DataChangeListenerSupport(this); - protected Shard(ShardIdentifier name, Map peerAddresses, - DatastoreContext datastoreContext, SchemaContext schemaContext) { - super(name.toString(), mapPeerAddresses(peerAddresses), - Optional.of(datastoreContext.getShardRaftConfig())); - this.name = name; - this.datastoreContext = datastoreContext; - this.schemaContext = schemaContext; + private ShardSnapshot restoreFromSnapshot; - String setting = System.getProperty("shard.persistent"); + private final ShardTransactionMessageRetrySupport messageRetrySupport; - this.persistent = !"false".equals(setting); + private final FrontendMetadata frontendMetadata = new FrontendMetadata(); - LOG.info("Shard created : {} persistent : {}", name, persistent); + protected Shard(final AbstractBuilder builder) { + super(builder.getId().toString(), builder.getPeerAddresses(), + Optional.of(builder.getDatastoreContext().getShardRaftConfig()), DataStoreVersions.CURRENT_VERSION); - store = InMemoryDOMDataStoreFactory.create(name.toString(), null, - datastoreContext.getDataStoreProperties()); + this.name = builder.getId().toString(); + this.datastoreContext = builder.getDatastoreContext(); + this.restoreFromSnapshot = builder.getRestoreFromSnapshot(); - if(schemaContext != null) { - store.onGlobalContextUpdated(schemaContext); + setPersistence(datastoreContext.isPersistent()); + + LOG.info("Shard created : {}, persistent : {}", name, datastoreContext.isPersistent()); + + ShardDataTreeChangeListenerPublisherActorProxy treeChangeListenerPublisher = + new ShardDataTreeChangeListenerPublisherActorProxy(getContext(), name + "-DTCL-publisher"); + ShardDataChangeListenerPublisherActorProxy dataChangeListenerPublisher = + new ShardDataChangeListenerPublisherActorProxy(getContext(), name + "-DCL-publisher"); + if (builder.getDataTree() != null) { + store = new ShardDataTree(this, builder.getSchemaContext(), builder.getDataTree(), + treeChangeListenerPublisher, dataChangeListenerPublisher, name); + } else { + store = new ShardDataTree(this, builder.getSchemaContext(), builder.getTreeType(), + treeChangeListenerPublisher, dataChangeListenerPublisher, name); } shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(), datastoreContext.getDataStoreMXBeanType()); - shardMBean.setDataStoreExecutor(store.getDomStoreExecutor()); - shardMBean.setNotificationManager(store.getDataChangeListenerNotificationManager()); + shardMBean.setShard(this); if (isMetricsCaptureEnabled()) { getContext().become(new MeteringBehavior(this)); } - } - private static Map mapPeerAddresses( - Map peerAddresses) { - Map map = new HashMap<>(); + commitCoordinator = new ShardCommitCoordinator(store, LOG, this.name); - for (Map.Entry entry : peerAddresses - .entrySet()) { - map.put(entry.getKey().toString(), entry.getValue()); - } + setTransactionCommitTimeout(); - return map; - } + // create a notifier actor for each cluster member + roleChangeNotifier = createRoleChangeNotifier(name.toString()); - public static Props props(final ShardIdentifier name, - final Map peerAddresses, - DatastoreContext datastoreContext, SchemaContext schemaContext) { - Preconditions.checkNotNull(name, "name should not be null"); - Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null"); - Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null"); - Preconditions.checkNotNull(schemaContext, "schemaContext should not be null"); + appendEntriesReplyTracker = new MessageTracker(AppendEntriesReply.class, + getRaftActorContext().getConfigParams().getIsolatedCheckIntervalInMillis()); - return Props.create(new ShardCreator(name, peerAddresses, datastoreContext, schemaContext)); - } + transactionActorFactory = new ShardTransactionActorFactory(store, datastoreContext, + new Dispatchers(context().system().dispatchers()).getDispatcherPath( + Dispatchers.DispatcherType.Transaction), self(), getContext(), shardMBean); - @Override public void onReceiveRecover(Object message) { - if(LOG.isDebugEnabled()) { - LOG.debug("onReceiveRecover: Received message {} from {}", - message.getClass().toString(), - getSender()); - } + snapshotCohort = ShardSnapshotCohort.create(getContext(), builder.getId().getMemberName(), store, LOG, + this.name); - if (message instanceof RecoveryFailure){ - LOG.error(((RecoveryFailure) message).cause(), "Recovery failed because of this cause"); - } else { - super.onReceiveRecover(message); - } - } - - @Override public void onReceiveCommand(Object message) { - if(LOG.isDebugEnabled()) { - LOG.debug("onReceiveCommand: Received message {} from {}", - message.getClass().toString(), - getSender()); - } - - if(message.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) { - // This must be for install snapshot. Don't want to open this up and trigger - // deSerialization - self() - .tell(new CaptureSnapshotReply(ReadDataReply.getNormalizedNodeByteString(message)), - self()); - - createSnapshotTransaction = null; - // Send a PoisonPill instead of sending close transaction because we do not really need - // a response - getSender().tell(PoisonPill.getInstance(), self()); - - } else if (message.getClass().equals(CloseTransactionChain.SERIALIZABLE_CLASS)){ - closeTransactionChain(CloseTransactionChain.fromSerializable(message)); - } else if (message instanceof RegisterChangeListener) { - registerChangeListener((RegisterChangeListener) message); - } else if (message instanceof UpdateSchemaContext) { - updateSchemaContext((UpdateSchemaContext) message); - } else if (message instanceof ForwardedCommitTransaction) { - handleForwardedCommit((ForwardedCommitTransaction) message); - } else if (message.getClass() - .equals(CreateTransaction.SERIALIZABLE_CLASS)) { - if (isLeader()) { - createTransaction(CreateTransaction.fromSerializable(message)); - } else if (getLeader() != null) { - getLeader().forward(message, getContext()); - } else { - getSender().tell(new akka.actor.Status.Failure(new IllegalStateException( - "Could not find shard leader so transaction cannot be created. This typically happens" + - " when system is coming up or recovering and a leader is being elected. Try again" + - " later.")), getSelf()); - } - } else if (message instanceof PeerAddressResolved) { - PeerAddressResolved resolved = (PeerAddressResolved) message; - setPeerAddress(resolved.getPeerId().toString(), - resolved.getPeerAddress()); - } else { - super.onReceiveCommand(message); - } + messageRetrySupport = new ShardTransactionMessageRetrySupport(this); } - private void closeTransactionChain(CloseTransactionChain closeTransactionChain) { - DOMStoreTransactionChain chain = - transactionChains.remove(closeTransactionChain.getTransactionChainId()); + private void setTransactionCommitTimeout() { + transactionCommitTimeout = TimeUnit.MILLISECONDS.convert( + datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS) / 2; + } - if(chain != null) { - chain.close(); - } + private Optional createRoleChangeNotifier(final String shardId) { + ActorRef shardRoleChangeNotifier = this.getContext().actorOf( + RoleChangeNotifier.getProps(shardId), shardId + "-notifier"); + return Optional.of(shardRoleChangeNotifier); } - private ActorRef createTypedTransactionActor( - int transactionType, - ShardTransactionIdentifier transactionId, - String transactionChainId ) { + @Override + public void postStop() { + LOG.info("Stopping Shard {}", persistenceId()); - DOMStoreTransactionFactory factory = store; + super.postStop(); - if(!transactionChainId.isEmpty()) { - factory = transactionChains.get(transactionChainId); - if(factory == null){ - DOMStoreTransactionChain transactionChain = store.createTransactionChain(); - transactionChains.put(transactionChainId, transactionChain); - factory = transactionChain; - } - } + messageRetrySupport.close(); - if(this.schemaContext == null){ - throw new NullPointerException("schemaContext should not be null"); + if (txCommitTimeoutCheckSchedule != null) { + txCommitTimeoutCheckSchedule.cancel(); } - if (transactionType - == TransactionProxy.TransactionType.READ_ONLY.ordinal()) { + commitCoordinator.abortPendingTransactions("Transaction aborted due to shutdown.", this); - shardMBean.incrementReadOnlyTransactionCount(); + shardMBean.unregisterMBean(); + } - return getContext().actorOf( - ShardTransaction.props(factory.newReadOnlyTransaction(), getSelf(), - schemaContext,datastoreContext, shardMBean), transactionId.toString()); + @Override + protected void handleRecover(final Object message) { + LOG.debug("{}: onReceiveRecover: Received message {} from {}", persistenceId(), message.getClass(), + getSender()); - } else if (transactionType - == TransactionProxy.TransactionType.READ_WRITE.ordinal()) { + super.handleRecover(message); + if (LOG.isTraceEnabled()) { + appendEntriesReplyTracker.begin(); + } + } - shardMBean.incrementReadWriteTransactionCount(); + @Override + protected void handleNonRaftCommand(final Object message) { + try (final MessageTracker.Context context = appendEntriesReplyTracker.received(message)) { + final Optional maybeError = context.error(); + if (maybeError.isPresent()) { + LOG.trace("{} : AppendEntriesReply failed to arrive at the expected interval {}", persistenceId(), + maybeError.get()); + } - return getContext().actorOf( - ShardTransaction.props(factory.newReadWriteTransaction(), getSelf(), - schemaContext, datastoreContext, shardMBean), transactionId.toString()); + if (CreateTransaction.isSerializedType(message)) { + handleCreateTransaction(message); + } else if (message instanceof BatchedModifications) { + handleBatchedModifications((BatchedModifications)message); + } else if (message instanceof ForwardedReadyTransaction) { + handleForwardedReadyTransaction((ForwardedReadyTransaction) message); + } else if (message instanceof ReadyLocalTransaction) { + handleReadyLocalTransaction((ReadyLocalTransaction)message); + } else if (CanCommitTransaction.isSerializedType(message)) { + handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message)); + } else if (CommitTransaction.isSerializedType(message)) { + handleCommitTransaction(CommitTransaction.fromSerializable(message)); + } else if (AbortTransaction.isSerializedType(message)) { + handleAbortTransaction(AbortTransaction.fromSerializable(message)); + } else if (CloseTransactionChain.isSerializedType(message)) { + closeTransactionChain(CloseTransactionChain.fromSerializable(message)); + } else if (message instanceof RegisterChangeListener) { + changeSupport.onMessage((RegisterChangeListener) message, isLeader(), hasLeader()); + } else if (message instanceof RegisterDataTreeChangeListener) { + treeChangeSupport.onMessage((RegisterDataTreeChangeListener) message, isLeader(), hasLeader()); + } else if (message instanceof UpdateSchemaContext) { + updateSchemaContext((UpdateSchemaContext) message); + } else if (message instanceof PeerAddressResolved) { + PeerAddressResolved resolved = (PeerAddressResolved) message; + setPeerAddress(resolved.getPeerId().toString(), + resolved.getPeerAddress()); + } else if (TX_COMMIT_TIMEOUT_CHECK_MESSAGE.equals(message)) { + store.checkForExpiredTransactions(transactionCommitTimeout); + commitCoordinator.checkForExpiredTransactions(transactionCommitTimeout, this); + } else if (message instanceof DatastoreContext) { + onDatastoreContext((DatastoreContext)message); + } else if (message instanceof RegisterRoleChangeListener) { + roleChangeNotifier.get().forward(message, context()); + } else if (message instanceof FollowerInitialSyncUpStatus) { + shardMBean.setFollowerInitialSyncStatus(((FollowerInitialSyncUpStatus) message).isInitialSyncDone()); + context().parent().tell(message, self()); + } else if (GET_SHARD_MBEAN_MESSAGE.equals(message)) { + sender().tell(getShardMBean(), self()); + } else if (message instanceof GetShardDataTree) { + sender().tell(store.getDataTree(), self()); + } else if (message instanceof ServerRemoved) { + context().parent().forward(message, context()); + } else if (ShardTransactionMessageRetrySupport.TIMER_MESSAGE_CLASS.isInstance(message)) { + messageRetrySupport.onTimerMessage(message); + } else if (message instanceof DataTreeCohortActorRegistry.CohortRegistryCommand) { + store.processCohortRegistryCommand(getSender(), + (DataTreeCohortActorRegistry.CohortRegistryCommand) message); + } else { + super.handleNonRaftCommand(message); + } + } + } + private boolean hasLeader() { + return getLeaderId() != null; + } - } else if (transactionType - == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) { + public int getPendingTxCommitQueueSize() { + return store.getQueueSize(); + } - shardMBean.incrementWriteOnlyTransactionCount(); + public int getCohortCacheSize() { + return commitCoordinator.getCohortCacheSize(); + } - return getContext().actorOf( - ShardTransaction.props(factory.newWriteOnlyTransaction(), getSelf(), - schemaContext, datastoreContext, shardMBean), transactionId.toString()); - } else { - throw new IllegalArgumentException( - "Shard="+name + ":CreateTransaction message has unidentified transaction type=" - + transactionType); - } + @Override + protected Optional getRoleChangeNotifier() { + return roleChangeNotifier; } - private void createTransaction(CreateTransaction createTransaction) { - createTransaction(createTransaction.getTransactionType(), - createTransaction.getTransactionId(), createTransaction.getTransactionChainId()); + @Override + protected LeaderStateChanged newLeaderStateChanged(final String memberId, final String leaderId, + final short leaderPayloadVersion) { + return isLeader() ? new ShardLeaderStateChanged(memberId, leaderId, store.getDataTree(), leaderPayloadVersion) + : new ShardLeaderStateChanged(memberId, leaderId, leaderPayloadVersion); } - private ActorRef createTransaction(int transactionType, String remoteTransactionId, String transactionChainId) { + protected void onDatastoreContext(final DatastoreContext context) { + datastoreContext = context; - ShardTransactionIdentifier transactionId = - ShardTransactionIdentifier.builder() - .remoteTransactionId(remoteTransactionId) - .build(); - if(LOG.isDebugEnabled()) { - LOG.debug("Creating transaction : {} ", transactionId); - } - ActorRef transactionActor = - createTypedTransactionActor(transactionType, transactionId, transactionChainId); + setTransactionCommitTimeout(); - getSender() - .tell(new CreateTransactionReply( - Serialization.serializedActorPath(transactionActor), - remoteTransactionId).toSerializable(), - getSelf()); + setPersistence(datastoreContext.isPersistent()); - return transactionActor; + updateConfigParams(datastoreContext.getShardRaftConfig()); } - private void syncCommitTransaction(DOMStoreWriteTransaction transaction) - throws ExecutionException, InterruptedException { - DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready(); - commitCohort.preCommit().get(); - commitCohort.commit().get(); + boolean canSkipPayload() { + // If we do not have any followers and we are not using persistence we can apply modification to the state + // immediately + return !hasFollowers() && !persistence().isRecoveryApplicable(); } + // applyState() will be invoked once consensus is reached on the payload + void persistPayload(final TransactionIdentifier transactionId, final Payload payload) { + // We are faking the sender + persistData(self(), transactionId, payload); + } - private void commit(final ActorRef sender, Object serialized) { - Modification modification = MutableCompositeModification - .fromSerializable(serialized, schemaContext); - DOMStoreThreePhaseCommitCohort cohort = - modificationToCohort.remove(serialized); - if (cohort == null) { - // If there's no cached cohort then we must be applying replicated state. - commitWithNewTransaction(serialized); - return; - } - - if(sender == null) { - LOG.error("Commit failed. Sender cannot be null"); - return; + private void handleCommitTransaction(final CommitTransaction commit) { + if (isLeader()) { + commitCoordinator.handleCommit(commit.getTransactionID(), getSender(), this); + } else { + ActorSelection leader = getLeader(); + if (leader == null) { + messageRetrySupport.addMessageToRetry(commit, getSender(), + "Could not commit transaction " + commit.getTransactionID()); + } else { + LOG.debug("{}: Forwarding CommitTransaction to leader {}", persistenceId(), leader); + leader.forward(commit, getContext()); + } } + } - ListenableFuture future = cohort.commit(); + private void handleCanCommitTransaction(final CanCommitTransaction canCommit) { + LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionID()); - Futures.addCallback(future, new FutureCallback() { - @Override - public void onSuccess(Void v) { - sender.tell(new CommitTransactionReply().toSerializable(), getSelf()); - shardMBean.incrementCommittedTransactionCount(); - shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); + if (isLeader()) { + commitCoordinator.handleCanCommit(canCommit.getTransactionID(), getSender(), this); + } else { + ActorSelection leader = getLeader(); + if (leader == null) { + messageRetrySupport.addMessageToRetry(canCommit, getSender(), + "Could not canCommit transaction " + canCommit.getTransactionID()); + } else { + LOG.debug("{}: Forwarding CanCommitTransaction to leader {}", persistenceId(), leader); + leader.forward(canCommit, getContext()); } + } + } - @Override - public void onFailure(Throwable t) { - LOG.error(t, "An exception happened during commit"); - shardMBean.incrementFailedTransactionsCount(); - sender.tell(new akka.actor.Status.Failure(t), getSelf()); + @SuppressWarnings("checkstyle:IllegalCatch") + protected void handleBatchedModificationsLocal(final BatchedModifications batched, final ActorRef sender) { + try { + commitCoordinator.handleBatchedModifications(batched, sender, this); + } catch (Exception e) { + LOG.error("{}: Error handling BatchedModifications for Tx {}", persistenceId(), + batched.getTransactionID(), e); + sender.tell(new akka.actor.Status.Failure(e), getSelf()); + } + } + + private void handleBatchedModifications(final BatchedModifications batched) { + // This message is sent to prepare the modifications transaction directly on the Shard as an + // optimization to avoid the extra overhead of a separate ShardTransaction actor. On the last + // BatchedModifications message, the caller sets the ready flag in the message indicating + // modifications are complete. The reply contains the cohort actor path (this actor) for the caller + // to initiate the 3-phase commit. This also avoids the overhead of sending an additional + // ReadyTransaction message. + + // If we're not the leader then forward to the leader. This is a safety measure - we shouldn't + // normally get here if we're not the leader as the front-end (TransactionProxy) should determine + // the primary/leader shard. However with timing and caching on the front-end, there's a small + // window where it could have a stale leader during leadership transitions. + // + boolean isLeaderActive = isLeaderActive(); + if (isLeader() && isLeaderActive) { + handleBatchedModificationsLocal(batched, getSender()); + } else { + ActorSelection leader = getLeader(); + if (!isLeaderActive || leader == null) { + messageRetrySupport.addMessageToRetry(batched, getSender(), + "Could not commit transaction " + batched.getTransactionID()); + } else { + // If this is not the first batch and leadership changed in between batched messages, + // we need to reconstruct previous BatchedModifications from the transaction + // DataTreeModification, honoring the max batched modification count, and forward all the + // previous BatchedModifications to the new leader. + Collection newModifications = commitCoordinator + .createForwardedBatchedModifications(batched, + datastoreContext.getShardBatchedModificationCount()); + + LOG.debug("{}: Forwarding {} BatchedModifications to leader {}", persistenceId(), + newModifications.size(), leader); + + for (BatchedModifications bm : newModifications) { + leader.forward(bm, getContext()); + } } - }); - + } } - private void commitWithNewTransaction(Object modification) { - DOMStoreWriteTransaction tx = store.newWriteOnlyTransaction(); - MutableCompositeModification.fromSerializable(modification, schemaContext).apply(tx); - try { - syncCommitTransaction(tx); - shardMBean.incrementCommittedTransactionCount(); - } catch (InterruptedException | ExecutionException e) { - shardMBean.incrementFailedTransactionsCount(); - LOG.error(e, "Failed to commit"); + private boolean failIfIsolatedLeader(final ActorRef sender) { + if (isIsolatedLeader()) { + sender.tell(new akka.actor.Status.Failure(new NoShardLeaderException(String.format( + "Shard %s was the leader but has lost contact with all of its followers. Either all" + + " other follower nodes are down or this node is isolated by a network partition.", + persistenceId()))), getSelf()); + return true; } + + return false; } - private void handleForwardedCommit(ForwardedCommitTransaction message) { - Object serializedModification = - message.getModification().toSerializable(); + protected boolean isIsolatedLeader() { + return getRaftState() == RaftState.IsolatedLeader; + } - modificationToCohort - .put(serializedModification, message.getCohort()); + @SuppressWarnings("checkstyle:IllegalCatch") + private void handleReadyLocalTransaction(final ReadyLocalTransaction message) { + LOG.debug("{}: handleReadyLocalTransaction for {}", persistenceId(), message.getTransactionID()); - if (persistent) { - this.persistData(getSender(), "identifier", - new CompositeModificationPayload(serializedModification)); + boolean isLeaderActive = isLeaderActive(); + if (isLeader() && isLeaderActive) { + try { + commitCoordinator.handleReadyLocalTransaction(message, getSender(), this); + } catch (Exception e) { + LOG.error("{}: Error handling ReadyLocalTransaction for Tx {}", persistenceId(), + message.getTransactionID(), e); + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } } else { - this.commit(getSender(), serializedModification); + ActorSelection leader = getLeader(); + if (!isLeaderActive || leader == null) { + messageRetrySupport.addMessageToRetry(message, getSender(), + "Could not commit transaction " + message.getTransactionID()); + } else { + LOG.debug("{}: Forwarding ReadyLocalTransaction to leader {}", persistenceId(), leader); + message.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion()); + leader.forward(message, getContext()); + } } } - private void updateSchemaContext(UpdateSchemaContext message) { - this.schemaContext = message.getSchemaContext(); - updateSchemaContext(message.getSchemaContext()); - store.onGlobalContextUpdated(message.getSchemaContext()); + private void handleForwardedReadyTransaction(final ForwardedReadyTransaction forwardedReady) { + LOG.debug("{}: handleForwardedReadyTransaction for {}", persistenceId(), forwardedReady.getTransactionID()); + + boolean isLeaderActive = isLeaderActive(); + if (isLeader() && isLeaderActive) { + commitCoordinator.handleForwardedReadyTransaction(forwardedReady, getSender(), this); + } else { + ActorSelection leader = getLeader(); + if (!isLeaderActive || leader == null) { + messageRetrySupport.addMessageToRetry(forwardedReady, getSender(), + "Could not commit transaction " + forwardedReady.getTransactionID()); + } else { + LOG.debug("{}: Forwarding ForwardedReadyTransaction to leader {}", persistenceId(), leader); + + ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(forwardedReady.getTransactionID(), + forwardedReady.getTransaction().getSnapshot(), forwardedReady.isDoImmediateCommit()); + readyLocal.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion()); + leader.forward(readyLocal, getContext()); + } + } } - @VisibleForTesting void updateSchemaContext(SchemaContext schemaContext) { - store.onGlobalContextUpdated(schemaContext); + private void handleAbortTransaction(final AbortTransaction abort) { + doAbortTransaction(abort.getTransactionID(), getSender()); } - private void registerChangeListener( - RegisterChangeListener registerChangeListener) { + void doAbortTransaction(final TransactionIdentifier transactionID, final ActorRef sender) { + commitCoordinator.handleAbort(transactionID, sender, this); + } - if(LOG.isDebugEnabled()) { - LOG.debug("registerDataChangeListener for {}", registerChangeListener - .getPath()); + private void handleCreateTransaction(final Object message) { + if (isLeader()) { + createTransaction(CreateTransaction.fromSerializable(message)); + } else if (getLeader() != null) { + getLeader().forward(message, getContext()); + } else { + getSender().tell(new akka.actor.Status.Failure(new NoShardLeaderException( + "Could not create a shard transaction", persistenceId())), getSelf()); } + } + private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) { + store.closeTransactionChain(closeTransactionChain.getIdentifier()); + } - ActorSelection dataChangeListenerPath = getContext() - .system().actorSelection( - registerChangeListener.getDataChangeListenerPath()); - - - // Notify the listener if notifications should be enabled or not - // If this shard is the leader then it will enable notifications else - // it will not - dataChangeListenerPath - .tell(new EnableNotification(isLeader()), getSelf()); + @SuppressWarnings("checkstyle:IllegalCatch") + private void createTransaction(final CreateTransaction createTransaction) { + try { + if (TransactionType.fromInt(createTransaction.getTransactionType()) != TransactionType.READ_ONLY + && failIfIsolatedLeader(getSender())) { + return; + } - // Now store a reference to the data change listener so it can be notified - // at a later point if notifications should be enabled or disabled - dataChangeListeners.add(dataChangeListenerPath); + ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(), + createTransaction.getTransactionId()); - AsyncDataChangeListener> - listener = new DataChangeListenerProxy(schemaContext, dataChangeListenerPath); + getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), + createTransaction.getTransactionId(), createTransaction.getVersion()).toSerializable(), getSelf()); + } catch (Exception e) { + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } + } - ListenerRegistration>> - registration = store.registerChangeListener(registerChangeListener.getPath(), - listener, registerChangeListener.getScope()); - ActorRef listenerRegistration = - getContext().actorOf( - DataChangeListenerRegistration.props(registration)); + private ActorRef createTransaction(final int transactionType, final TransactionIdentifier transactionId) { + LOG.debug("{}: Creating transaction : {} ", persistenceId(), transactionId); + return transactionActorFactory.newShardTransaction(TransactionType.fromInt(transactionType), + transactionId); + } - if(LOG.isDebugEnabled()) { - LOG.debug( - "registerDataChangeListener sending reply, listenerRegistrationPath = {} " - , listenerRegistration.path().toString()); - } + private void updateSchemaContext(final UpdateSchemaContext message) { + updateSchemaContext(message.getSchemaContext()); + } - getSender() - .tell(new RegisterChangeListenerReply(listenerRegistration.path()), - getSelf()); + @VisibleForTesting + void updateSchemaContext(final SchemaContext schemaContext) { + store.updateSchemaContext(schemaContext); } - private boolean isMetricsCaptureEnabled(){ + private boolean isMetricsCaptureEnabled() { CommonConfig config = new CommonConfig(getContext().system().settings().config()); return config.isMetricCaptureEnabled(); } @Override - protected - void startLogRecoveryBatch(int maxBatchSize) { - currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize); - - if(LOG.isDebugEnabled()) { - LOG.debug("{} : starting log recovery batch with max size {}", persistenceId(), maxBatchSize); - } + @VisibleForTesting + public RaftActorSnapshotCohort getRaftActorSnapshotCohort() { + return snapshotCohort; } @Override - protected void appendRecoveredLogEntry(Payload data) { - if (data instanceof CompositeModificationPayload) { - currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification()); - } else { - LOG.error("Unknown state received {} during recovery", data); - } + @Nonnull + protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() { + return new ShardRecoveryCoordinator(store, + restoreFromSnapshot != null ? restoreFromSnapshot.getSnapshot() : null, persistenceId(), LOG); } @Override - protected void applyRecoverySnapshot(ByteString snapshot) { - if(recoveryCoordinator == null) { - recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext); - } + protected void onRecoveryComplete() { + restoreFromSnapshot = null; - recoveryCoordinator.submit(snapshot, store.newWriteOnlyTransaction()); + //notify shard manager + getContext().parent().tell(new ActorInitialized(), getSelf()); - if(LOG.isDebugEnabled()) { - LOG.debug("{} : submitted recovery sbapshot", persistenceId()); + // Being paranoid here - this method should only be called once but just in case... + if (txCommitTimeoutCheckSchedule == null) { + // Schedule a message to be periodically sent to check if the current in-progress + // transaction should be expired and aborted. + FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS); + txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule( + period, period, getSelf(), + TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender()); } } @Override - protected void applyCurrentLogRecoveryBatch() { - if(recoveryCoordinator == null) { - recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext); - } - - recoveryCoordinator.submit(currentLogRecoveryBatch, store.newWriteOnlyTransaction()); - - if(LOG.isDebugEnabled()) { - LOG.debug("{} : submitted log recovery batch with size {}", persistenceId(), - currentLogRecoveryBatch.size()); + protected void applyState(final ActorRef clientActor, final Identifier identifier, final Object data) { + if (data instanceof Payload) { + try { + store.applyReplicatedPayload(identifier, (Payload)data); + } catch (DataValidationFailedException | IOException e) { + LOG.error("{}: Error applying replica {}", persistenceId(), identifier, e); + } + } else { + LOG.error("{}: Unknown state for {} received {}", persistenceId(), identifier, data); } } @Override - protected void onRecoveryComplete() { - if(recoveryCoordinator != null) { - Collection txList = recoveryCoordinator.getTransactions(); + protected void onStateChanged() { + boolean isLeader = isLeader(); + boolean hasLeader = hasLeader(); + changeSupport.onLeadershipChange(isLeader, hasLeader); + treeChangeSupport.onLeadershipChange(isLeader, hasLeader); - if(LOG.isDebugEnabled()) { - LOG.debug("{} : recovery complete - committing {} Tx's", persistenceId(), txList.size()); + // If this actor is no longer the leader close all the transaction chains + if (!isLeader) { + if (LOG.isDebugEnabled()) { + LOG.debug( + "{}: onStateChanged: Closing all transaction chains because shard {} is no longer the leader", + persistenceId(), getId()); } - for(DOMStoreWriteTransaction tx: txList) { - try { - syncCommitTransaction(tx); - shardMBean.incrementCommittedTransactionCount(); - } catch (InterruptedException | ExecutionException e) { - shardMBean.incrementFailedTransactionsCount(); - LOG.error(e, "Failed to commit"); - } - } + store.closeAllTransactionChains(); } - recoveryCoordinator = null; - currentLogRecoveryBatch = null; - updateJournalStats(); - - //notify shard manager - getContext().parent().tell(new ActorInitialized(), getSelf()); + if (hasLeader && !isIsolatedLeader()) { + messageRetrySupport.retryMessages(); + } } @Override - protected void applyState(ActorRef clientActor, String identifier, Object data) { - - if (data instanceof CompositeModificationPayload) { - Object modification = ((CompositeModificationPayload) data).getModification(); - - if (modification != null) { - commit(clientActor, modification); + protected void onLeaderChanged(final String oldLeader, final String newLeader) { + shardMBean.incrementLeadershipChangeCount(); + + boolean hasLeader = hasLeader(); + if (hasLeader && !isLeader()) { + // Another leader was elected. If we were the previous leader and had pending transactions, convert + // them to transaction messages and send to the new leader. + ActorSelection leader = getLeader(); + if (leader != null) { + Collection messagesToForward = convertPendingTransactionsToMessages(); + + if (!messagesToForward.isEmpty()) { + LOG.debug("{}: Forwarding {} pending transaction messages to leader {}", persistenceId(), + messagesToForward.size(), leader); + + for (Object message : messagesToForward) { + leader.tell(message, self()); + } + } } else { - LOG.error( - "modification is null - this is very unexpected, clientActor = {}, identifier = {}", - identifier, clientActor != null ? clientActor.path().toString() : null); + commitCoordinator.abortPendingTransactions("The transacton was aborted due to inflight leadership " + + "change and the leader address isn't available.", this); } - - } else { - LOG.error("Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}", - data, data.getClass().getClassLoader(), - CompositeModificationPayload.class.getClassLoader()); } - updateJournalStats(); - + if (hasLeader && !isIsolatedLeader()) { + messageRetrySupport.retryMessages(); + } } - private void updateJournalStats() { - ReplicatedLogEntry lastLogEntry = getLastLogEntry(); - - if (lastLogEntry != null) { - shardMBean.setLastLogIndex(lastLogEntry.getIndex()); - shardMBean.setLastLogTerm(lastLogEntry.getTerm()); - } + /** + * Clears all pending transactions and converts them to messages to be forwarded to a new leader. + * + * @return the converted messages + */ + public Collection convertPendingTransactionsToMessages() { + return commitCoordinator.convertPendingTransactionsToMessages( + datastoreContext.getShardBatchedModificationCount()); + } - shardMBean.setCommitIndex(getCommitIndex()); - shardMBean.setLastApplied(getLastApplied()); + @Override + protected void pauseLeader(final Runnable operation) { + LOG.debug("{}: In pauseLeader, operation: {}", persistenceId(), operation); + store.setRunOnPendingTransactionsComplete(operation); } @Override - protected void createSnapshot() { - if (createSnapshotTransaction == null) { + public String persistenceId() { + return this.name; + } - // Create a transaction. We are really going to treat the transaction as a worker - // so that this actor does not get block building the snapshot - createSnapshotTransaction = createTransaction( - TransactionProxy.TransactionType.READ_ONLY.ordinal(), - "createSnapshot", ""); + @VisibleForTesting + ShardCommitCoordinator getCommitCoordinator() { + return commitCoordinator; + } - createSnapshotTransaction.tell( - new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self()); + public DatastoreContext getDatastoreContext() { + return datastoreContext; + } - } + @VisibleForTesting + public ShardDataTree getDataStore() { + return store; } @VisibleForTesting - @Override - protected void applySnapshot(ByteString snapshot) { - // Since this will be done only on Recovery or when this actor is a Follower - // we can safely commit everything in here. We not need to worry about event notifications - // as they would have already been disabled on the follower + ShardStats getShardMBean() { + return shardMBean; + } - LOG.info("Applying snapshot"); - try { - DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction(); - NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(snapshot); - NormalizedNode node = new NormalizedNodeToNodeCodec(schemaContext) - .decode(serializedNode); + public static Builder builder() { + return new Builder(); + } - // delete everything first - transaction.delete(YangInstanceIdentifier.builder().build()); + public abstract static class AbstractBuilder, S extends Shard> { + private final Class shardClass; + private ShardIdentifier id; + private Map peerAddresses = Collections.emptyMap(); + private DatastoreContext datastoreContext; + private SchemaContext schemaContext; + private DatastoreSnapshot.ShardSnapshot restoreFromSnapshot; + private TipProducingDataTree dataTree; + private volatile boolean sealed; - // Add everything from the remote node back - transaction.write(YangInstanceIdentifier.builder().build(), node); - syncCommitTransaction(transaction); - } catch (InvalidProtocolBufferException | InterruptedException | ExecutionException e) { - LOG.error(e, "An exception occurred when applying snapshot"); - } finally { - LOG.info("Done applying snapshot"); + protected AbstractBuilder(final Class shardClass) { + this.shardClass = shardClass; } - } - @Override protected void onStateChanged() { - for (ActorSelection dataChangeListener : dataChangeListeners) { - dataChangeListener - .tell(new EnableNotification(isLeader()), getSelf()); + protected void checkSealed() { + Preconditions.checkState(!sealed, "Builder isalready sealed - further modifications are not allowed"); } - shardMBean.setRaftState(getRaftState().name()); - shardMBean.setCurrentTerm(getCurrentTerm()); - - // If this actor is no longer the leader close all the transaction chains - if(!isLeader()){ - for(Map.Entry entry : transactionChains.entrySet()){ - if(LOG.isDebugEnabled()) { - LOG.debug( - "onStateChanged: Closing transaction chain {} because shard {} is no longer the leader", - entry.getKey(), getId()); - } - entry.getValue().close(); - } + @SuppressWarnings("unchecked") + private T self() { + return (T) this; + } - transactionChains.clear(); + public T id(final ShardIdentifier newId) { + checkSealed(); + this.id = newId; + return self(); } - } - @Override protected void onLeaderChanged(String oldLeader, String newLeader) { - shardMBean.setLeader(newLeader); - } + public T peerAddresses(final Map newPeerAddresses) { + checkSealed(); + this.peerAddresses = newPeerAddresses; + return self(); + } - @Override public String persistenceId() { - return this.name.toString(); - } + public T datastoreContext(final DatastoreContext newDatastoreContext) { + checkSealed(); + this.datastoreContext = newDatastoreContext; + return self(); + } - private static class ShardCreator implements Creator { + public T schemaContext(final SchemaContext newSchemaContext) { + checkSealed(); + this.schemaContext = newSchemaContext; + return self(); + } - private static final long serialVersionUID = 1L; + public T restoreFromSnapshot(final DatastoreSnapshot.ShardSnapshot newRestoreFromSnapshot) { + checkSealed(); + this.restoreFromSnapshot = newRestoreFromSnapshot; + return self(); + } - final ShardIdentifier name; - final Map peerAddresses; - final DatastoreContext datastoreContext; - final SchemaContext schemaContext; + public T dataTree(final TipProducingDataTree newDataTree) { + checkSealed(); + this.dataTree = newDataTree; + return self(); + } - ShardCreator(ShardIdentifier name, Map peerAddresses, - DatastoreContext datastoreContext, SchemaContext schemaContext) { - this.name = name; - this.peerAddresses = peerAddresses; - this.datastoreContext = datastoreContext; - this.schemaContext = schemaContext; + public ShardIdentifier getId() { + return id; } - @Override - public Shard create() throws Exception { - return new Shard(name, peerAddresses, datastoreContext, schemaContext); + public Map getPeerAddresses() { + return peerAddresses; } - } - @VisibleForTesting - NormalizedNode readStore(YangInstanceIdentifier id) - throws ExecutionException, InterruptedException { - DOMStoreReadTransaction transaction = store.newReadOnlyTransaction(); + public DatastoreContext getDatastoreContext() { + return datastoreContext; + } - CheckedFuture>, ReadFailedException> future = - transaction.read(id); + public SchemaContext getSchemaContext() { + return schemaContext; + } - Optional> optional = future.get(); - NormalizedNode node = optional.isPresent()? optional.get() : null; + public DatastoreSnapshot.ShardSnapshot getRestoreFromSnapshot() { + return restoreFromSnapshot; + } - transaction.close(); + public TipProducingDataTree getDataTree() { + return dataTree; + } - return node; - } + public TreeType getTreeType() { + switch (datastoreContext.getLogicalStoreType()) { + case CONFIGURATION: + return TreeType.CONFIGURATION; + case OPERATIONAL: + return TreeType.OPERATIONAL; + default: + throw new IllegalStateException("Unhandled logical store type " + + datastoreContext.getLogicalStoreType()); + } + } - @VisibleForTesting - void writeToStore(YangInstanceIdentifier id, NormalizedNode node) - throws ExecutionException, InterruptedException { - DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction(); + protected void verify() { + Preconditions.checkNotNull(id, "id should not be null"); + Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null"); + Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null"); + Preconditions.checkNotNull(schemaContext, "schemaContext should not be null"); + } - transaction.write(id, node); + public Props props() { + sealed = true; + verify(); + return Props.create(shardClass, this); + } + } - syncCommitTransaction(transaction); + public static class Builder extends AbstractBuilder { + private Builder() { + super(Shard.class); + } } - @VisibleForTesting - ShardStats getShardMBean() { - return shardMBean; + Ticker ticker() { + return Ticker.systemTicker(); } }