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=23ad747d4cf3d4c7247271c09f5c64fa9fc44c2e;hb=d4fa6758d6b94aad894854c0fe6fcd82e7bbefd6;hp=1978c5559b0ee13e60bc6b70b0d3a5ad568bddb1;hpb=f86f7e8c204fb19615c45e669a764c623576e1a3;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 1978c5559b..23ad747d4c 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 @@ -24,6 +24,8 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import javax.annotation.Nonnull; 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.ShardCommitCoordinator.CohortEntry; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; @@ -53,7 +55,6 @@ import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContex import org.opendaylight.controller.cluster.datastore.modification.Modification; import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification; import org.opendaylight.controller.cluster.datastore.utils.Dispatchers; -import org.opendaylight.controller.cluster.datastore.utils.MessageTracker; import org.opendaylight.controller.cluster.notifications.LeaderStateChanged; import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener; import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier; @@ -66,7 +67,6 @@ 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.CompositeModificationByteStringPayload; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException; import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType; @@ -83,10 +83,21 @@ import scala.concurrent.duration.FiniteDuration; */ public class Shard extends RaftActor { - protected static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = "txCommitTimeoutCheck"; + @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 = "getShardMBeanMessage"; + 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"; @@ -208,16 +219,14 @@ public class Shard extends RaftActor { } @Override - protected void handleCommand(final Object message) { - - MessageTracker.Context context = appendEntriesReplyTracker.received(message); - - if(context.error().isPresent()){ - LOG.trace("{} : AppendEntriesReply failed to arrive at the expected interval {}", persistenceId(), - context.error()); - } + 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()); + } - try { if (CreateTransaction.isSerializedType(message)) { handleCreateTransaction(message); } else if (message instanceof BatchedModifications) { @@ -244,7 +253,7 @@ public class Shard extends RaftActor { PeerAddressResolved resolved = (PeerAddressResolved) message; setPeerAddress(resolved.getPeerId().toString(), resolved.getPeerAddress()); - } else if (message.equals(TX_COMMIT_TIMEOUT_CHECK_MESSAGE)) { + } else if (TX_COMMIT_TIMEOUT_CHECK_MESSAGE.equals(message)) { commitCoordinator.checkForExpiredTransactions(transactionCommitTimeout, this); } else if(message instanceof DatastoreContext) { onDatastoreContext((DatastoreContext)message); @@ -261,11 +270,12 @@ public class Shard extends RaftActor { context().parent().forward(message, context()); } else if(ShardTransactionMessageRetrySupport.TIMER_MESSAGE_CLASS.isInstance(message)) { messageRetrySupport.onTimerMessage(message); + } else if (message instanceof DataTreeCohortActorRegistry.CohortRegistryCommand) { + commitCoordinator.processCohortRegistryCommand(getSender(), + (DataTreeCohortActorRegistry.CohortRegistryCommand) message); } else { - super.handleCommand(message); + super.handleNonRaftCommand(message); } - } finally { - context.done(); } } @@ -288,9 +298,8 @@ public class Shard extends RaftActor { @Override protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId, short leaderPayloadVersion) { - return new ShardLeaderStateChanged(memberId, leaderId, - isLeader() ? Optional.of(store.getDataTree()) : Optional.absent(), - leaderPayloadVersion); + return isLeader() ? new ShardLeaderStateChanged(memberId, leaderId, store.getDataTree(), leaderPayloadVersion) + : new ShardLeaderStateChanged(memberId, leaderId, leaderPayloadVersion); } protected void onDatastoreContext(DatastoreContext context) { @@ -329,9 +338,9 @@ public class Shard extends RaftActor { private void handleCommitTransaction(final CommitTransaction commit) { if (isLeader()) { - if(!commitCoordinator.handleCommit(commit.getTransactionID(), getSender(), this)) { - shardMBean.incrementFailedTransactionsCount(); - } + if(!commitCoordinator.handleCommit(commit.getTransactionID(), getSender(), this)) { + shardMBean.incrementFailedTransactionsCount(); + } } else { ActorSelection leader = getLeader(); if (leader == null) { @@ -349,7 +358,7 @@ public class Shard extends RaftActor { try { try { - cohortEntry.commit(); + cohortEntry.commit(); } catch(ExecutionException e) { // We may get a "store tree and candidate base differ" IllegalStateException from commit under // certain edge case scenarios so we'll try to re-apply the candidate from scratch as a last @@ -427,7 +436,7 @@ public class Shard extends RaftActor { LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionID()); if (isLeader()) { - commitCoordinator.handleCanCommit(canCommit.getTransactionID(), getSender(), this); + commitCoordinator.handleCanCommit(canCommit.getTransactionID(), getSender(), this); } else { ActorSelection leader = getLeader(); if (leader == null) { @@ -442,7 +451,7 @@ public class Shard extends RaftActor { protected void handleBatchedModificationsLocal(BatchedModifications batched, ActorRef sender) { try { - commitCoordinator.handleBatchedModifications(batched, sender, this); + commitCoordinator.handleBatchedModifications(batched, sender, this, store.getSchemaContext()); } catch (Exception e) { LOG.error("{}: Error handling BatchedModifications for Tx {}", persistenceId(), batched.getTransactionID(), e); @@ -511,7 +520,7 @@ public class Shard extends RaftActor { boolean isLeaderActive = isLeaderActive(); if (isLeader() && isLeaderActive) { try { - commitCoordinator.handleReadyLocalTransaction(message, getSender(), this); + commitCoordinator.handleReadyLocalTransaction(message, getSender(), this, store.getSchemaContext()); } catch (Exception e) { LOG.error("{}: Error handling ReadyLocalTransaction for Tx {}", persistenceId(), message.getTransactionID(), e); @@ -535,7 +544,8 @@ public class Shard extends RaftActor { boolean isLeaderActive = isLeaderActive(); if (isLeader() && isLeaderActive) { - commitCoordinator.handleForwardedReadyTransaction(forwardedReady, getSender(), this); + commitCoordinator.handleForwardedReadyTransaction(forwardedReady, getSender(), this, + store.getSchemaContext()); } else { ActorSelection leader = getLeader(); if (!isLeaderActive || leader == null) {