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
index 7f5bec1ff6c25dd7dd0650217e327e96a82c9e5c..83ba07b69ac8f6e3ed253d18e420315065069e4d 100644 (file)
@@ -8,6 +8,7 @@
 package org.opendaylight.controller.cluster.databroker.actors.dds;
 
 import akka.actor.ActorRef;
+import com.google.common.base.MoreObjects;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
@@ -17,12 +18,17 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import java.util.ArrayDeque;
 import java.util.Deque;
+import java.util.Iterator;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Consumer;
 import javax.annotation.Nonnull;
 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;
@@ -30,8 +36,9 @@ import org.opendaylight.controller.cluster.access.commands.TransactionCommitSucc
 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
+import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
-import org.opendaylight.controller.cluster.access.concepts.RequestException;
+import org.opendaylight.controller.cluster.access.concepts.Request;
 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
 import org.opendaylight.controller.cluster.access.concepts.Response;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
@@ -73,26 +80,81 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    private enum SealState {
-        /**
-         * The user has not sealed the transaction yet.
-         */
-        OPEN,
-        /**
-         * The user has sealed the transaction, but has not issued a canCommit().
-         */
-        SEALED,
-        /**
-         * The user has sealed the transaction and has issued a canCommit().
-         */
-        FLUSHED,
+    // Generic state base class. Direct instances are used for fast paths, sub-class is used for successor transitions
+    private static class State {
+        private final String string;
+
+        State(final String string) {
+            this.string = Preconditions.checkNotNull(string);
+        }
+
+        @Override
+        public final String toString() {
+            return string;
+        }
     }
 
-    private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
+    // State class used when a successor has interfered. Contains coordinator latch, the successor and previous state
+    private static final class SuccessorState extends State {
+        private final CountDownLatch latch = new CountDownLatch(1);
+        private AbstractProxyTransaction successor;
+        private State prevState;
+
+        SuccessorState() {
+            super("successor");
+        }
+
+        // Synchronize with succession process and return the successor
+        AbstractProxyTransaction await() {
+            try {
+                latch.await();
+            } catch (InterruptedException e) {
+                LOG.warn("Interrupted while waiting for latch of {}", successor);
+                throw Throwables.propagate(e);
+            }
+            return successor;
+        }
+
+        void finish() {
+            latch.countDown();
+        }
+
+        State getPrevState() {
+            return prevState;
+        }
 
+        void setPrevState(final State prevState) {
+            Verify.verify(this.prevState == null);
+            this.prevState = Preconditions.checkNotNull(prevState);
+        }
+
+        // To be called from safe contexts, where successor is known to be completed
+        AbstractProxyTransaction getSuccessor() {
+            return Verify.verifyNotNull(successor);
+        }
+
+        void setSuccessor(final AbstractProxyTransaction successor) {
+            Verify.verify(this.successor == null);
+            this.successor = Preconditions.checkNotNull(successor);
+        }
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
+    private static final AtomicIntegerFieldUpdater<AbstractProxyTransaction> SEALED_UPDATER =
+            AtomicIntegerFieldUpdater.newUpdater(AbstractProxyTransaction.class, "sealed");
+    private static final AtomicReferenceFieldUpdater<AbstractProxyTransaction, State> STATE_UPDATER =
+            AtomicReferenceFieldUpdater.newUpdater(AbstractProxyTransaction.class, State.class, "state");
+    private static final State OPEN = new State("open");
+    private static final State SEALED = new State("sealed");
+    private static final State FLUSHED = new State("flushed");
+
+    // Touched from client actor thread only
     private final Deque<Object> successfulRequests = new ArrayDeque<>();
     private final ProxyHistory parent;
 
+    // Accessed from user thread only, which may not access this object concurrently
+    private long sequence;
+
     /*
      * Atomic state-keeping is required to synchronize the process of propagating completed transaction state towards
      * the backend -- which may include a successor.
@@ -105,17 +167,20 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * or timeout, when a successor is injected.
      *
      * This leaves the problem of needing to completely transferring state just after all queued messages are replayed
-     * after a successor was injected, so that it can be properly sealed if we are racing.
+     * after a successor was injected, so that it can be properly sealed if we are racing. Further complication comes
+     * from lock ordering, where the successor injection works with a locked queue and locks proxy objects -- leading
+     * to a potential AB-BA deadlock in case of a naive implementation.
+     *
+     * For tracking user-visible state we use a single volatile int, which is flipped atomically from 0 to 1 exactly
+     * once in {@link AbstractProxyTransaction#seal()}. That keeps common operations fast, as they need to perform
+     * only a single volatile read to assert state correctness.
+     *
+     * For synchronizing client actor (successor-injecting) and user (commit-driving) thread, we keep a separate state
+     * variable. It uses pre-allocated objects for fast paths (i.e. no successor present) and a per-transition object
+     * for slow paths (when successor is injected/present).
      */
-    private volatile SealState sealed = SealState.OPEN;
-    @GuardedBy("this")
-    private AbstractProxyTransaction successor;
-    @GuardedBy("this")
-    private CountDownLatch successorLatch;
-
-    // Accessed from user thread only, which may not access this object concurrently
-    private long sequence;
-
+    private volatile int sealed = 0;
+    private volatile State state = OPEN;
 
     AbstractProxyTransaction(final ProxyHistory parent) {
         this.parent = Preconditions.checkNotNull(parent);
@@ -137,16 +202,19 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     final void delete(final YangInstanceIdentifier path) {
+        checkReadWrite();
         checkNotSealed();
         doDelete(path);
     }
 
     final void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+        checkReadWrite();
         checkNotSealed();
         doMerge(path, data);
     }
 
     final void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+        checkReadWrite();
         checkNotSealed();
         doWrite(path, data);
     }
@@ -170,50 +238,53 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * Seal this transaction before it is either committed or aborted.
      */
     final void seal() {
-        final CountDownLatch localLatch;
-
-        synchronized (this) {
-            checkNotSealed();
-            doSeal();
-
-            // Fast path: no successor
-            if (successor == null) {
-                sealed = SealState.SEALED;
-                parent.onTransactionSealed(this);
-                return;
-            }
+        // Transition user-visible state first
+        final boolean success = SEALED_UPDATER.compareAndSet(this, 0, 1);
+        Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
+        internalSeal();
+    }
 
-            localLatch = successorLatch;
+    final void ensureSealed() {
+        if (SEALED_UPDATER.compareAndSet(this, 0, 1)) {
+            internalSeal();
         }
+    }
 
-        // Slow path: wait for the latch
-        LOG.debug("{} waiting on successor latch", getIdentifier());
-        try {
-            localLatch.await();
-        } catch (InterruptedException e) {
-            LOG.warn("{} interrupted while waiting for latch", getIdentifier());
-            throw Throwables.propagate(e);
-        }
+    private void internalSeal() {
+        doSeal();
+        parent.onTransactionSealed(this);
 
-        synchronized (this) {
-            LOG.debug("{} reacquired lock", getIdentifier());
+        // Now deal with state transfer, which can occur via successor or a follow-up canCommit() or directCommit().
+        if (!STATE_UPDATER.compareAndSet(this, OPEN, SEALED)) {
+            // Slow path: wait for the successor to complete
+            final AbstractProxyTransaction successor = awaitSuccessor();
 
+            // At this point the successor has completed transition and is possibly visible by the user thread, which is
+            // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
+            // Propagate state and seal the successor.
             flushState(successor);
-            successor.seal();
-
-            sealed = SealState.FLUSHED;
-            parent.onTransactionSealed(this);
+            successor.ensureSealed();
         }
     }
 
     private void checkNotSealed() {
-        Preconditions.checkState(sealed == SealState.OPEN, "Transaction %s has already been sealed", getIdentifier());
+        Preconditions.checkState(sealed == 0, "Transaction %s has already been sealed", getIdentifier());
+    }
+
+    private void checkSealed() {
+        Preconditions.checkState(sealed != 0, "Transaction %s has not been sealed yet", getIdentifier());
+    }
+
+    private SuccessorState getSuccessorState() {
+        final State local = state;
+        Verify.verify(local instanceof SuccessorState, "State %s has unexpected class", local);
+        return (SuccessorState) local;
     }
 
-    private SealState checkSealed() {
-        final SealState local = sealed;
-        Preconditions.checkState(local != SealState.OPEN, "Transaction %s has not been sealed yet", getIdentifier());
-        return local;
+    private void checkReadWrite() {
+        if (isSnapshotOnly()) {
+            throw new UnsupportedOperationException("Transaction " + getIdentifier() + " is a read-only snapshot");
+        }
     }
 
     final void recordSuccessfulRequest(final @Nonnull TransactionRequest<?> req) {
@@ -246,14 +317,14 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             if (t instanceof TransactionAbortSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
-                ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
             // This is a terminal request, hence we do not need to record it
             LOG.debug("Transaction {} abort completed", this);
-            parent.completeTransaction(this);
+            purge();
         });
     }
 
@@ -268,73 +339,49 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @return Future completion
      */
     final ListenableFuture<Boolean> directCommit() {
-        final CountDownLatch localLatch;
+        checkReadWrite();
+        checkSealed();
 
+        // Precludes startReconnect() from interfering with the fast path
         synchronized (this) {
-            final SealState local = checkSealed();
-
-            // Fast path: no successor asserted
-            if (successor == null) {
-                Verify.verify(local == SealState.SEALED);
-
+            if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
                 final SettableFuture<Boolean> ret = SettableFuture.create();
                 sendRequest(Verify.verifyNotNull(commitRequest(false)), t -> {
                     if (t instanceof TransactionCommitSuccess) {
                         ret.set(Boolean.TRUE);
                     } else if (t instanceof RequestFailure) {
-                        ret.setException(((RequestFailure<?, ?>) t).getCause());
+                        ret.setException(((RequestFailure<?, ?>) t).getCause().unwrap());
                     } else {
                         ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
                     }
 
                     // This is a terminal request, hence we do not need to record it
                     LOG.debug("Transaction {} directCommit completed", this);
-                    parent.completeTransaction(this);
+                    purge();
                 });
 
-                sealed = SealState.FLUSHED;
                 return ret;
             }
-
-            // We have a successor, take its latch
-            localLatch = successorLatch;
         }
 
-        // Slow path: we need to wait for the successor to completely propagate
-        LOG.debug("{} waiting on successor latch", getIdentifier());
-        try {
-            localLatch.await();
-        } catch (InterruptedException e) {
-            LOG.warn("{} interrupted while waiting for latch", getIdentifier());
-            throw Throwables.propagate(e);
-        }
-
-        synchronized (this) {
-            LOG.debug("{} reacquired lock", getIdentifier());
-
-            final SealState local = sealed;
-            Verify.verify(local == SealState.FLUSHED);
-
-            return successor.directCommit();
-        }
+        // We have had some interference with successor injection, wait for it to complete and defer to the successor.
+        return awaitSuccessor().directCommit();
     }
 
     final void canCommit(final VotingFuture<?> ret) {
-        final CountDownLatch localLatch;
+        checkReadWrite();
+        checkSealed();
 
+        // Precludes startReconnect() from interfering with the fast path
         synchronized (this) {
-            final SealState local = checkSealed();
-
-            // Fast path: no successor asserted
-            if (successor == null) {
-                Verify.verify(local == SealState.SEALED);
-
+            if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
                 final TransactionRequest<?> req = Verify.verifyNotNull(commitRequest(true));
+
                 sendRequest(req, t -> {
                     if (t instanceof TransactionCanCommitSuccess) {
                         ret.voteYes();
                     } else if (t instanceof RequestFailure) {
-                        ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                        ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
                     } else {
                         ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
                     }
@@ -343,34 +390,20 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     LOG.debug("Transaction {} canCommit completed", this);
                 });
 
-                sealed = SealState.FLUSHED;
                 return;
             }
-
-            // We have a successor, take its latch
-            localLatch = successorLatch;
-        }
-
-        // Slow path: we need to wait for the successor to completely propagate
-        LOG.debug("{} waiting on successor latch", getIdentifier());
-        try {
-            localLatch.await();
-        } catch (InterruptedException e) {
-            LOG.warn("{} interrupted while waiting for latch", getIdentifier());
-            throw Throwables.propagate(e);
         }
 
-        synchronized (this) {
-            LOG.debug("{} reacquired lock", getIdentifier());
-
-            final SealState local = sealed;
-            Verify.verify(local == SealState.FLUSHED);
+        // We have had some interference with successor injection, wait for it to complete and defer to the successor.
+        awaitSuccessor().canCommit(ret);
+    }
 
-            successor.canCommit(ret);
-        }
+    private AbstractProxyTransaction awaitSuccessor() {
+        return getSuccessorState().await();
     }
 
     final void preCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
         final TransactionRequest<?> req = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
@@ -379,41 +412,87 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             if (t instanceof TransactionPreCommitSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
-                ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
-            recordSuccessfulRequest(req);
-            LOG.debug("Transaction {} preCommit completed", this);
+            onPreCommitComplete(req);
         });
     }
 
-    void doCommit(final VotingFuture<?> ret) {
+    private void onPreCommitComplete(final TransactionRequest<?> req) {
+        /*
+         * The backend has agreed that the transaction has entered PRE_COMMIT phase, meaning it will be committed
+         * to storage after the timeout completes.
+         *
+         * All state has been replicated to the backend, hence we do not need to keep it around. Retain only
+         * the precommit request, so we know which request to use for resync.
+         */
+        LOG.debug("Transaction {} preCommit completed, clearing successfulRequests", this);
+        successfulRequests.clear();
+
+        // TODO: this works, but can contain some useless state (like batched operations). Create an empty
+        //       equivalent of this request and store that.
+        recordSuccessfulRequest(req);
+    }
+
+    final void doCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
         sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
             if (t instanceof TransactionCommitSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
-                ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
             LOG.debug("Transaction {} doCommit completed", this);
+            purge();
+        });
+    }
+
+    void purge() {
+        successfulRequests.clear();
+
+        final TransactionRequest<?> req = new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
+        sendRequest(req, t -> {
+            LOG.debug("Transaction {} purge completed", this);
             parent.completeTransaction(this);
         });
     }
 
-    final synchronized void startReconnect(final AbstractProxyTransaction successor) {
-        Preconditions.checkState(this.successor == null);
-        this.successor = Preconditions.checkNotNull(successor);
+    // 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
+        // state. This method is called with the queue still unlocked.
+        final SuccessorState nextState = new SuccessorState();
+        final State prevState = STATE_UPDATER.getAndSet(this, nextState);
+
+        LOG.debug("Start reconnect of proxy {} previous state {}", this, prevState);
+        Verify.verify(!(prevState instanceof SuccessorState), "Proxy %s duplicate reconnect attempt after %s", this,
+            prevState);
+
+        // We have asserted a slow-path state, seal(), canCommit(), directCommit() are forced to slow paths, which will
+        // wait until we unblock nextState's latch before accessing state. Now we record prevState for later use and we
+        // are done.
+        nextState.setPrevState(prevState);
+    }
+
+    // Called with the connection locked
+    final void replayMessages(final AbstractProxyTransaction successor,
+            final Iterable<ConnectionEntry> enqueuedEntries) {
+        final SuccessorState local = getSuccessorState();
+        local.setSuccessor(successor);
 
+        // Replay successful requests first
         for (Object obj : successfulRequests) {
             if (obj instanceof TransactionRequest) {
-                LOG.debug("Forwarding request {} to successor {}", obj, successor);
-                successor.handleForwardedRemoteRequest((TransactionRequest<?>) obj, null);
+                LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
+                successor.replay((TransactionRequest<?>) obj, response -> { });
             } else {
                 Verify.verify(obj instanceof IncrementSequence);
                 successor.incrementSequence(((IncrementSequence) obj).getDelta());
@@ -422,28 +501,58 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
         successfulRequests.clear();
 
+        // Now replay whatever is in the connection
+        final Iterator<ConnectionEntry> it = enqueuedEntries.iterator();
+        while (it.hasNext()) {
+            final ConnectionEntry e = it.next();
+            final Request<?, ?> req = e.getRequest();
+
+            if (getIdentifier().equals(req.getTarget())) {
+                Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
+                LOG.debug("Forwarding queued request {} to successor {}", req, successor);
+                successor.replay((TransactionRequest<?>) req, e.getCallback());
+                it.remove();
+            }
+        }
+
         /*
-         * Before releasing the lock we need to make sure that a call to seal() blocks until we have completed
-         * finishConnect().
+         * Check the state at which we have started the reconnect attempt. State transitions triggered while we were
+         * reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
+         * at the end of this method.
          */
-        successorLatch = new CountDownLatch(1);
+        final State prevState = local.getPrevState();
+        if (SEALED.equals(prevState)) {
+            LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
+            flushState(successor);
+            successor.ensureSealed();
+        }
     }
 
-    final synchronized void finishReconnect() {
-        Preconditions.checkState(successorLatch != null);
-
-        if (sealed == SealState.SEALED) {
-            /*
-             * If this proxy is in the 'sealed, have not sent canCommit' state. If so, we need to forward current
-             * leftover state to the successor now.
-             */
-            flushState(successor);
-            successor.seal();
-            sealed = SealState.FLUSHED;
+    /**
+     * 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();
+        LOG.debug("Finishing reconnect of proxy {}", this);
 
-        // All done, release the latch, unblocking seal() and canCommit()
-        successorLatch.countDown();
+        // All done, release the latch, unblocking seal() and canCommit() slow paths
+        local.finish();
     }
 
     /**
@@ -452,11 +561,9 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      *
      * @param request Request to be forwarded
      * @param callback Original callback
-     * @throws RequestException when the request is unhandled by the successor
      */
-    final synchronized void replayRequest(final TransactionRequest<?> request,
-            final Consumer<Response<?, ?>> callback) {
-        Preconditions.checkState(successor != null, "%s does not have a successor set", this);
+    final void forwardRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
+        final AbstractProxyTransaction successor = getSuccessorState().getSuccessor();
 
         if (successor instanceof LocalProxyTransaction) {
             forwardToLocal((LocalProxyTransaction)successor, request, callback);
@@ -467,16 +574,17 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    abstract void doDelete(final YangInstanceIdentifier path);
+    abstract boolean isSnapshotOnly();
 
-    abstract void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
+    abstract void doDelete(YangInstanceIdentifier path);
 
-    abstract void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
+    abstract void doMerge(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 
-    abstract CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path);
+    abstract void doWrite(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 
-    abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(
-            final YangInstanceIdentifier path);
+    abstract CheckedFuture<Boolean, ReadFailedException> doExists(YangInstanceIdentifier path);
+
+    abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(YangInstanceIdentifier path);
 
     abstract void doSeal();
 
@@ -488,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.
@@ -498,18 +616,23 @@ 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.
+     * 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 forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
-            Consumer<Response<?, ?>> callback);
+    abstract void handleForwardedRemoteRequest(TransactionRequest<?> request,
+            @Nullable 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);
+    @Override
+    public final String toString() {
+        return MoreObjects.toStringHelper(this).add("identifier", getIdentifier()).add("state", state).toString();
+    }
 }