BUG-8403: move successor allocation to AbstractProxyTransaction

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / AbstractProxyTransaction.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
index b8057c5afe428fd162e0ddf02c6a2c6d11eab108..f4f2e19f76745685f04abdd190e3a79e07b32291 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
@@ -334,7 +334,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
         sendRequest(abortRequest(), resp -> {
             LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
-            sendPurge();
+            enqueuePurge();
         });
     }
 
@@ -352,7 +352,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
             // This is a terminal request, hence we do not need to record it
             LOG.debug("Transaction {} abort completed", this);
-            sendPurge();
+            enqueuePurge();
         });
     }
 
@@ -403,7 +403,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
                     // This is a terminal request, hence we do not need to record it
                     LOG.debug("Transaction {} directCommit completed", this);
-                    sendPurge();
+                    enqueuePurge();
                 });
 
                 return ret;
@@ -497,20 +497,27 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             }
 
             LOG.debug("Transaction {} doCommit completed", this);
-            sendPurge();
+            enqueuePurge();
         });
     }
 
-    private void sendPurge() {
-        sendPurge(null);
+    private void enqueuePurge() {
+        enqueuePurge(null);
     }
 
-    final void sendPurge(final Consumer<Response<?, ?>> callback) {
-        sendRequest(purgeRequest(), resp -> completePurge(resp, callback));
+    final void enqueuePurge(final Consumer<Response<?, ?>> callback) {
+        // Purge request are dispatched internally, hence should not wait
+        enqueuePurge(callback, parent.currentTime());
     }
 
     final void enqueuePurge(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
-        enqueueRequest(purgeRequest(), resp -> completePurge(resp, callback), enqueuedTicks);
+        enqueueRequest(purgeRequest(), resp -> {
+            LOG.debug("Transaction {} purge completed", this);
+            parent.completeTransaction(this);
+            if (callback != null) {
+                callback.accept(resp);
+            }
+        }, enqueuedTicks);
     }
 
     private TransactionPurgeRequest purgeRequest() {
@@ -518,14 +525,6 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         return new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
     }
 
-    private void completePurge(final Response<?, ?> resp, final Consumer<Response<?, ?>> callback) {
-        LOG.debug("Transaction {} purge completed", this);
-        parent.completeTransaction(this);
-        if (callback != null) {
-            callback.accept(resp);
-        }
-    }
-
     // Called with the connection unlocked
     final synchronized void startReconnect() {
         // At this point canCommit/directCommit are blocked, we assert a new successor state, retrieving the previous
@@ -544,9 +543,13 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     // Called with the connection locked
-    final void replayMessages(final AbstractProxyTransaction successor,
-            final Iterable<ConnectionEntry> enqueuedEntries) {
+    final void replayMessages(final ProxyHistory successorHistory, final Iterable<ConnectionEntry> enqueuedEntries) {
         final SuccessorState local = getSuccessorState();
+        final State prevState = local.getPrevState();
+
+        final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
+            isSnapshotOnly());
+        LOG.debug("{} created successor transaction proxy {}", this, successor);
         local.setSuccessor(successor);
 
         // Replay successful requests first
@@ -594,7 +597,6 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
          * reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
          * at the end of this method.
          */
-        final State prevState = local.getPrevState();
         if (SEALED.equals(prevState)) {
             LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
             flushState(successor);