X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=72b31d5598d80aaf7f7961e33c562d72dbbb7bc1;hb=4f64d27f5434ed7bc82de2b6f6a57ba13ccaec4b;hp=74e25323bc711434dac93efe71594defc761f3f0;hpb=0281535ab08fd795e42df66d25e9a904ff941ad7;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 74e25323bc..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);
 
@@ -228,7 +219,7 @@ public class Shard extends RaftActor {
         try {
             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);
@@ -240,7 +231,7 @@ public class Shard extends RaftActor {
                 handleCommitTransaction(CommitTransaction.fromSerializable(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();
@@ -437,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<BatchedModifications> newModifications = commitCoordinator.createForwardedBatchedModifications(
+                        batched, datastoreContext.getShardBatchedModificationCount());
+
+                LOG.debug("{}: Forwarding {} BatchedModifications to leader {}", persistenceId(),
+                        newModifications.size(), leader);
+
+                for(BatchedModifications bm: newModifications) {
+                    leader.forward(bm, getContext());
+                }
             }
         }
     }
@@ -533,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) {
@@ -548,8 +546,7 @@ 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(), createTransaction.getVersion()).toSerializable(), getSelf());
@@ -559,7 +556,7 @@ public class Shard extends RaftActor {
     }
 
     private ActorRef createTransaction(int transactionType, String remoteTransactionId,
-            String transactionChainId, short clientVersion) {
+            String transactionChainId) {
 
 
         ShardTransactionIdentifier transactionId = new ShardTransactionIdentifier(remoteTransactionId);
@@ -569,7 +566,7 @@ public class Shard extends RaftActor {
         }
 
         ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId,
-                transactionChainId, clientVersion);
+                transactionChainId);
 
         return transactionActor;
     }