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=7165581d8103fc64148e745e477bc1ec7f2278d6;hb=refs%2Fchanges%2F87%2F38587%2F3;hp=ab878055ae31324fc5a616de9b858da1f24a2b6d;hpb=93c17c55ca0c3690d77c96cdfed91b6f186fe56a;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 ab878055ae..7165581d81 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 @@ -25,8 +25,8 @@ 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; import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier; import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier; @@ -66,7 +66,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 +82,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 +218,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 +252,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 +269,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 +297,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 +337,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 +357,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 +435,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 +450,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 +519,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 +543,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) {