BUG-8372: fix AbstractProxyTransaction.replayMessages()

[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 86d7f237411f0068fc4649371d85245c993c8ddf..83ba07b69ac8f6e3ed253d18e420315065069e4d 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
@@ -28,6 +28,7 @@ import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
+import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
@@ -491,7 +492,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         for (Object obj : successfulRequests) {
             if (obj instanceof TransactionRequest) {
                 LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
-                successor.handleForwardedRemoteRequest((TransactionRequest<?>) obj, response -> { });
+                successor.replay((TransactionRequest<?>) obj, response -> { });
             } else {
                 Verify.verify(obj instanceof IncrementSequence);
                 successor.incrementSequence(((IncrementSequence) obj).getDelta());
@@ -508,8 +509,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
             if (getIdentifier().equals(req.getTarget())) {
                 Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
-                LOG.debug("Forwarding queued request{} to successor {}", req, successor);
-                successor.handleForwardedRemoteRequest((TransactionRequest<?>) req, e.getCallback());
+                LOG.debug("Forwarding queued request {} to successor {}", req, successor);
+                successor.replay((TransactionRequest<?>) req, e.getCallback());
                 it.remove();
             }
         }
@@ -527,6 +528,24 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
+    /**
+     * Invoked from {@link #replayMessages(AbstractProxyTransaction, Iterable)} to have successor adopt an in-flight
+     * request.
+     *
+     * <p>
+     * Note: this method is invoked by the predecessor on the successor.
+     *
+     * @param request Request which needs to be forwarded
+     * @param callback Callback to be invoked once the request completes
+     */
+    private void replay(TransactionRequest<?> request, Consumer<Response<?, ?>> callback) {
+        if (request instanceof AbstractLocalTransactionRequest) {
+            handleForwardedLocalRequest((AbstractLocalTransactionRequest<?>) request, callback);
+        } else {
+            handleForwardedRemoteRequest(request, callback);
+        }
+    }
+
     // Called with the connection locked
     final void finishReconnect() {
         final SuccessorState local = getSuccessorState();
@@ -543,7 +562,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @param request Request to be forwarded
      * @param callback Original callback
      */
-    final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
+    final void forwardRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
         final AbstractProxyTransaction successor = getSuccessorState().getSuccessor();
 
         if (successor instanceof LocalProxyTransaction) {
@@ -577,9 +596,19 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     abstract TransactionRequest<?> commitRequest(boolean coordinated);
 
     /**
-     * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor. There is
-     * no equivalent of this call from {@link LocalProxyTransaction} because it does not send a request until all
-     * operations are packaged in the message.
+     * Replay a request originating in this proxy to a successor remote proxy.
+     */
+    abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
+            Consumer<Response<?, ?>> callback);
+
+    /**
+     * Replay a request originating in this proxy to a successor local proxy.
+     */
+    abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
+            Consumer<Response<?, ?>> callback);
+
+    /**
+     * Invoked from {@link LocalProxyTransaction} when it replays its successful requests to its successor.
      *
      * <p>
      * Note: this method is invoked by the predecessor on the successor.
@@ -587,20 +616,20 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @param request Request which needs to be forwarded
      * @param callback Callback to be invoked once the request completes
      */
-    abstract void handleForwardedRemoteRequest(TransactionRequest<?> request,
+    abstract void handleForwardedLocalRequest(AbstractLocalTransactionRequest<?> request,
             @Nullable Consumer<Response<?, ?>> callback);
 
     /**
-     * Replay a request originating in this proxy to a successor remote proxy.
-     */
-    abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
-            Consumer<Response<?, ?>> callback);
-
-    /**
-     * Replay a request originating in this proxy to a successor local proxy.
+     * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor.
+     *
+     * <p>
+     * Note: this method is invoked by the predecessor on the successor.
+     *
+     * @param request Request which needs to be forwarded
+     * @param callback Callback to be invoked once the request completes
      */
-    abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
-            Consumer<Response<?, ?>> callback);
+    abstract void handleForwardedRemoteRequest(TransactionRequest<?> request,
+            @Nullable Consumer<Response<?, ?>> callback);
 
     @Override
     public final String toString() {