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=72b31d5598d80aaf7f7961e33c562d72dbbb7bc1;hb=4f64d27f5434ed7bc82de2b6f6a57ba13ccaec4b;hp=9ea22fc1967f117ec8284ffbaee140da27c394d9;hpb=7ec296abdea94bbdc336276731ee545a2fb13e7c;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 9ea22fc196..72b31d5598 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 @@ -12,12 +12,12 @@ import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.Cancellable; import akka.actor.Props; -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 java.io.IOException; +import java.util.Collection; import java.util.Collections; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -134,7 +134,9 @@ public class Shard extends RaftActor { LOG.info("Shard created : {}, persistent : {}", name, datastoreContext.isPersistent()); - store = new ShardDataTree(builder.getSchemaContext(), builder.getTreeType()); + store = new ShardDataTree(builder.getSchemaContext(), builder.getTreeType(), + new ShardDataTreeChangeListenerPublisherActorProxy(getContext(), name + "-DTCL-publisher"), + new ShardDataChangeListenerPublisherActorProxy(getContext(), name + "-DCL-publisher"), name); shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(), datastoreContext.getDataStoreMXBeanType()); @@ -194,29 +196,18 @@ public class Shard extends RaftActor { } @Override - public void onReceiveRecover(final Object message) throws Exception { - if(LOG.isDebugEnabled()) { - LOG.debug("{}: onReceiveRecover: Received message {} from {}", persistenceId(), - message.getClass().toString(), getSender()); - } - - if (message instanceof RecoveryFailure){ - LOG.error("{}: Recovery failed because of this cause", - persistenceId(), ((RecoveryFailure) message).cause()); + protected void handleRecover(final Object message) { + LOG.debug("{}: onReceiveRecover: Received message {} from {}", persistenceId(), message.getClass(), + getSender()); - // Even though recovery failed, we still need to finish our recovery, eg send the - // ActorInitialized message and start the txCommitTimeoutCheckSchedule. - onRecoveryComplete(); - } else { - super.onReceiveRecover(message); - if(LOG.isTraceEnabled()) { - appendEntriesReplyTracker.begin(); - } + super.handleRecover(message); + if (LOG.isTraceEnabled()) { + appendEntriesReplyTracker.begin(); } } @Override - public void onReceiveCommand(final Object message) throws Exception { + protected void handleCommand(final Object message) { MessageTracker.Context context = appendEntriesReplyTracker.received(message); @@ -226,21 +217,21 @@ public class Shard extends RaftActor { } try { - if (CreateTransaction.SERIALIZABLE_CLASS.isInstance(message)) { + if (CreateTransaction.isSerializedType(message)) { handleCreateTransaction(message); - } else if (BatchedModifications.class.isInstance(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.SERIALIZABLE_CLASS.isInstance(message)) { + } else if (CanCommitTransaction.isSerializedType(message)) { handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message)); - } else if (CommitTransaction.SERIALIZABLE_CLASS.isInstance(message)) { + } else if (CommitTransaction.isSerializedType(message)) { handleCommitTransaction(CommitTransaction.fromSerializable(message)); - } else if (AbortTransaction.SERIALIZABLE_CLASS.isInstance(message)) { + } else if (AbortTransaction.isSerializedType(message)) { handleAbortTransaction(AbortTransaction.fromSerializable(message)); - } else if (CloseTransactionChain.SERIALIZABLE_CLASS.isInstance(message)) { + } else if (CloseTransactionChain.isSerializedType(message)) { closeTransactionChain(CloseTransactionChain.fromSerializable(message)); } else if (message instanceof RegisterChangeListener) { changeSupport.onMessage((RegisterChangeListener) message, isLeader(), hasLeader()); @@ -270,7 +261,7 @@ public class Shard extends RaftActor { } else if(ShardTransactionMessageRetrySupport.TIMER_MESSAGE_CLASS.isInstance(message)) { messageRetrySupport.onTimerMessage(message); } else { - super.onReceiveCommand(message); + super.handleCommand(message); } } finally { context.done(); @@ -347,7 +338,7 @@ public class Shard extends RaftActor { try { cohortEntry.commit(); - sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf()); + sender.tell(CommitTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), getSelf()); shardMBean.incrementCommittedTransactionCount(); shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); @@ -384,7 +375,8 @@ public class Shard extends RaftActor { LOG.error("{}: Failed to re-apply transaction {}", persistenceId(), transactionID, e); } - sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf()); + sender.tell(CommitTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), + getSelf()); } else { // This really shouldn't happen - it likely means that persistence or replication // took so long to complete such that the cohort entry was expired from the cache. @@ -436,11 +428,19 @@ public class Shard extends RaftActor { messageRetrySupport.addMessageToRetry(batched, getSender(), "Could not commit transaction " + batched.getTransactionID()); } else { - // TODO: what if this is not the first batch and leadership changed in between batched messages? - // We could check if the commitCoordinator already has a cached entry and forward all the previous - // batched modifications. - LOG.debug("{}: Forwarding BatchedModifications to leader {}", persistenceId(), leader); - leader.forward(batched, getContext()); + // 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()); + } } } } @@ -532,11 +532,10 @@ public class Shard extends RaftActor { } private ActorRef createTypedTransactionActor(int transactionType, - ShardTransactionIdentifier transactionId, String transactionChainId, - short clientVersion ) { + ShardTransactionIdentifier transactionId, String transactionChainId) { return transactionActorFactory.newShardTransaction(TransactionType.fromInt(transactionType), - transactionId, transactionChainId, clientVersion); + transactionId, transactionChainId); } private void createTransaction(CreateTransaction createTransaction) { @@ -547,18 +546,17 @@ public class Shard extends RaftActor { } ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(), - createTransaction.getTransactionId(), createTransaction.getTransactionChainId(), - createTransaction.getVersion()); + createTransaction.getTransactionId(), createTransaction.getTransactionChainId()); getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), - createTransaction.getTransactionId()).toSerializable(), getSelf()); + createTransaction.getTransactionId(), createTransaction.getVersion()).toSerializable(), getSelf()); } catch (Exception e) { getSender().tell(new akka.actor.Status.Failure(e), getSelf()); } } private ActorRef createTransaction(int transactionType, String remoteTransactionId, - String transactionChainId, short clientVersion) { + String transactionChainId) { ShardTransactionIdentifier transactionId = new ShardTransactionIdentifier(remoteTransactionId); @@ -568,7 +566,7 @@ public class Shard extends RaftActor { } ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId, - transactionChainId, clientVersion); + transactionChainId); return transactionActor; }