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

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.databroker.actors.dds;
   9
  10 import akka.actor.ActorRef;
  11 import com.google.common.base.MoreObjects;
  12 import com.google.common.base.Optional;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.base.Throwables;
  15 import com.google.common.base.Verify;
  16 import com.google.common.util.concurrent.CheckedFuture;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import com.google.common.util.concurrent.SettableFuture;
  19 import java.util.ArrayDeque;
  20 import java.util.Deque;
  21 import java.util.Iterator;
  22 import java.util.concurrent.CountDownLatch;
  23 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
  24 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
  25 import java.util.function.Consumer;
  26 import javax.annotation.Nonnull;
  27 import javax.annotation.Nullable;
  28 import javax.annotation.concurrent.GuardedBy;
  29 import javax.annotation.concurrent.NotThreadSafe;
  30 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
  31 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
  32 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
  33 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
  34 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
  35 import org.opendaylight.controller.cluster.access.commands.TransactionCommitSuccess;
  36 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
  37 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
  38 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
  39 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
  40 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
  41 import org.opendaylight.controller.cluster.access.concepts.Request;
  42 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
  43 import org.opendaylight.controller.cluster.access.concepts.Response;
  44 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  45 import org.opendaylight.mdsal.common.api.ReadFailedException;
  46 import org.opendaylight.yangtools.concepts.Identifiable;
  47 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  48 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51
  52 /**
  53  * Class translating transaction operations towards a particular backend shard.
  54  *
  55  * <p>
  56  * This class is not safe to access from multiple application threads, as is usual for transactions. Internal state
  57  * transitions coming from interactions with backend are expected to be thread-safe.
  58  *
  59  * <p>
  60  * This class interacts with the queueing mechanism in ClientActorBehavior, hence once we arrive at a decision
  61  * to use either a local or remote implementation, we are stuck with it. We can re-evaluate on the next transaction.
  62  *
  63  * @author Robert Varga
  64  */
  65 abstract class AbstractProxyTransaction implements Identifiable<TransactionIdentifier> {
  66     /**
  67      * Marker object used instead of read-type of requests, which are satisfied only once. This has a lower footprint
  68      * and allows compressing multiple requests into a single entry.
  69      */
  70     @NotThreadSafe
  71     private static final class IncrementSequence {
  72         private long delta = 1;
  73
  74         long getDelta() {
  75             return delta;
  76         }
  77
  78         void incrementDelta() {
  79             delta++;
  80         }
  81     }
  82
  83     // Generic state base class. Direct instances are used for fast paths, sub-class is used for successor transitions
  84     private static class State {
  85         private final String string;
  86
  87         State(final String string) {
  88             this.string = Preconditions.checkNotNull(string);
  89         }
  90
  91         @Override
  92         public final String toString() {
  93             return string;
  94         }
  95     }
  96
  97     // State class used when a successor has interfered. Contains coordinator latch, the successor and previous state
  98     private static final class SuccessorState extends State {
  99         private final CountDownLatch latch = new CountDownLatch(1);
 100         private AbstractProxyTransaction successor;
 101         private State prevState;
 102
 103         SuccessorState() {
 104             super("successor");
 105         }
 106
 107         // Synchronize with succession process and return the successor
 108         AbstractProxyTransaction await() {
 109             try {
 110                 latch.await();
 111             } catch (InterruptedException e) {
 112                 LOG.warn("Interrupted while waiting for latch of {}", successor);
 113                 throw Throwables.propagate(e);
 114             }
 115             return successor;
 116         }
 117
 118         void finish() {
 119             latch.countDown();
 120         }
 121
 122         State getPrevState() {
 123             return prevState;
 124         }
 125
 126         void setPrevState(final State prevState) {
 127             Verify.verify(this.prevState == null);
 128             this.prevState = Preconditions.checkNotNull(prevState);
 129         }
 130
 131         // To be called from safe contexts, where successor is known to be completed
 132         AbstractProxyTransaction getSuccessor() {
 133             return Verify.verifyNotNull(successor);
 134         }
 135
 136         void setSuccessor(final AbstractProxyTransaction successor) {
 137             Verify.verify(this.successor == null);
 138             this.successor = Preconditions.checkNotNull(successor);
 139         }
 140     }
 141
 142     private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
 143     private static final AtomicIntegerFieldUpdater<AbstractProxyTransaction> SEALED_UPDATER =
 144             AtomicIntegerFieldUpdater.newUpdater(AbstractProxyTransaction.class, "sealed");
 145     private static final AtomicReferenceFieldUpdater<AbstractProxyTransaction, State> STATE_UPDATER =
 146             AtomicReferenceFieldUpdater.newUpdater(AbstractProxyTransaction.class, State.class, "state");
 147     private static final State OPEN = new State("open");
 148     private static final State SEALED = new State("sealed");
 149     private static final State FLUSHED = new State("flushed");
 150
 151     // Touched from client actor thread only
 152     private final Deque<Object> successfulRequests = new ArrayDeque<>();
 153     private final ProxyHistory parent;
 154
 155     // Accessed from user thread only, which may not access this object concurrently
 156     private long sequence;
 157
 158     /*
 159      * Atomic state-keeping is required to synchronize the process of propagating completed transaction state towards
 160      * the backend -- which may include a successor.
 161      *
 162      * Successor, unlike {@link AbstractProxyTransaction#seal()} is triggered from the client actor thread, which means
 163      * the successor placement needs to be atomic with regard to the application thread.
 164      *
 165      * In the common case, the application thread performs performs the seal operations and then "immediately" sends
 166      * the corresponding message. The uncommon case is when the seal and send operations race with a connect completion
 167      * or timeout, when a successor is injected.
 168      *
 169      * This leaves the problem of needing to completely transferring state just after all queued messages are replayed
 170      * after a successor was injected, so that it can be properly sealed if we are racing. Further complication comes
 171      * from lock ordering, where the successor injection works with a locked queue and locks proxy objects -- leading
 172      * to a potential AB-BA deadlock in case of a naive implementation.
 173      *
 174      * For tracking user-visible state we use a single volatile int, which is flipped atomically from 0 to 1 exactly
 175      * once in {@link AbstractProxyTransaction#seal()}. That keeps common operations fast, as they need to perform
 176      * only a single volatile read to assert state correctness.
 177      *
 178      * For synchronizing client actor (successor-injecting) and user (commit-driving) thread, we keep a separate state
 179      * variable. It uses pre-allocated objects for fast paths (i.e. no successor present) and a per-transition object
 180      * for slow paths (when successor is injected/present).
 181      */
 182     private volatile int sealed = 0;
 183     private volatile State state = OPEN;
 184
 185     AbstractProxyTransaction(final ProxyHistory parent) {
 186         this.parent = Preconditions.checkNotNull(parent);
 187     }
 188
 189     final ActorRef localActor() {
 190         return parent.localActor();
 191     }
 192
 193     private void incrementSequence(final long delta) {
 194         sequence += delta;
 195         LOG.debug("Transaction {} incremented sequence to {}", this, sequence);
 196     }
 197
 198     final long nextSequence() {
 199         final long ret = sequence++;
 200         LOG.debug("Transaction {} allocated sequence {}", this, ret);
 201         return ret;
 202     }
 203
 204     final void delete(final YangInstanceIdentifier path) {
 205         checkReadWrite();
 206         checkNotSealed();
 207         doDelete(path);
 208     }
 209
 210     final void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 211         checkReadWrite();
 212         checkNotSealed();
 213         doMerge(path, data);
 214     }
 215
 216     final void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 217         checkReadWrite();
 218         checkNotSealed();
 219         doWrite(path, data);
 220     }
 221
 222     final CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
 223         checkNotSealed();
 224         return doExists(path);
 225     }
 226
 227     final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
 228         checkNotSealed();
 229         return doRead(path);
 230     }
 231
 232     final void sendRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
 233         LOG.debug("Transaction proxy {} sending request {} callback {}", this, request, callback);
 234         parent.sendRequest(request, callback);
 235     }
 236
 237     /**
 238      * Seal this transaction before it is either committed or aborted.
 239      */
 240     final void seal() {
 241         // Transition user-visible state first
 242         final boolean success = SEALED_UPDATER.compareAndSet(this, 0, 1);
 243         Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
 244         internalSeal();
 245     }
 246
 247     final void ensureSealed() {
 248         if (SEALED_UPDATER.compareAndSet(this, 0, 1)) {
 249             internalSeal();
 250         }
 251     }
 252
 253     private void internalSeal() {
 254         doSeal();
 255         parent.onTransactionSealed(this);
 256
 257         // Now deal with state transfer, which can occur via successor or a follow-up canCommit() or directCommit().
 258         if (!STATE_UPDATER.compareAndSet(this, OPEN, SEALED)) {
 259             // Slow path: wait for the successor to complete
 260             final AbstractProxyTransaction successor = awaitSuccessor();
 261
 262             // At this point the successor has completed transition and is possibly visible by the user thread, which is
 263             // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
 264             // Propagate state and seal the successor.
 265             flushState(successor);
 266             successor.ensureSealed();
 267         }
 268     }
 269
 270     private void checkNotSealed() {
 271         Preconditions.checkState(sealed == 0, "Transaction %s has already been sealed", getIdentifier());
 272     }
 273
 274     private void checkSealed() {
 275         Preconditions.checkState(sealed != 0, "Transaction %s has not been sealed yet", getIdentifier());
 276     }
 277
 278     private SuccessorState getSuccessorState() {
 279         final State local = state;
 280         Verify.verify(local instanceof SuccessorState, "State %s has unexpected class", local);
 281         return (SuccessorState) local;
 282     }
 283
 284     private void checkReadWrite() {
 285         if (isSnapshotOnly()) {
 286             throw new UnsupportedOperationException("Transaction " + getIdentifier() + " is a read-only snapshot");
 287         }
 288     }
 289
 290     final void recordSuccessfulRequest(final @Nonnull TransactionRequest<?> req) {
 291         successfulRequests.add(Verify.verifyNotNull(req));
 292     }
 293
 294     final void recordFinishedRequest() {
 295         final Object last = successfulRequests.peekLast();
 296         if (last instanceof IncrementSequence) {
 297             ((IncrementSequence) last).incrementDelta();
 298         } else {
 299             successfulRequests.addLast(new IncrementSequence());
 300         }
 301     }
 302
 303     /**
 304      * Abort this transaction. This is invoked only for read-only transactions and will result in an explicit message
 305      * being sent to the backend.
 306      */
 307     final void abort() {
 308         checkNotSealed();
 309         doAbort();
 310         parent.abortTransaction(this);
 311     }
 312
 313     final void abort(final VotingFuture<Void> ret) {
 314         checkSealed();
 315
 316         sendAbort(t -> {
 317             if (t instanceof TransactionAbortSuccess) {
 318                 ret.voteYes();
 319             } else if (t instanceof RequestFailure) {
 320                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
 321             } else {
 322                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
 323             }
 324
 325             // This is a terminal request, hence we do not need to record it
 326             LOG.debug("Transaction {} abort completed", this);
 327             purge();
 328         });
 329     }
 330
 331     final void sendAbort(final Consumer<Response<?, ?>> callback) {
 332         sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback);
 333     }
 334
 335     /**
 336      * Commit this transaction, possibly in a coordinated fashion.
 337      *
 338      * @param coordinated True if this transaction should be coordinated across multiple participants.
 339      * @return Future completion
 340      */
 341     final ListenableFuture<Boolean> directCommit() {
 342         checkReadWrite();
 343         checkSealed();
 344
 345         // Precludes startReconnect() from interfering with the fast path
 346         synchronized (this) {
 347             if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
 348                 final SettableFuture<Boolean> ret = SettableFuture.create();
 349                 sendRequest(Verify.verifyNotNull(commitRequest(false)), t -> {
 350                     if (t instanceof TransactionCommitSuccess) {
 351                         ret.set(Boolean.TRUE);
 352                     } else if (t instanceof RequestFailure) {
 353                         ret.setException(((RequestFailure<?, ?>) t).getCause().unwrap());
 354                     } else {
 355                         ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
 356                     }
 357
 358                     // This is a terminal request, hence we do not need to record it
 359                     LOG.debug("Transaction {} directCommit completed", this);
 360                     purge();
 361                 });
 362
 363                 return ret;
 364             }
 365         }
 366
 367         // We have had some interference with successor injection, wait for it to complete and defer to the successor.
 368         return awaitSuccessor().directCommit();
 369     }
 370
 371     final void canCommit(final VotingFuture<?> ret) {
 372         checkReadWrite();
 373         checkSealed();
 374
 375         // Precludes startReconnect() from interfering with the fast path
 376         synchronized (this) {
 377             if (STATE_UPDATER.compareAndSet(this, SEALED, FLUSHED)) {
 378                 final TransactionRequest<?> req = Verify.verifyNotNull(commitRequest(true));
 379
 380                 sendRequest(req, t -> {
 381                     if (t instanceof TransactionCanCommitSuccess) {
 382                         ret.voteYes();
 383                     } else if (t instanceof RequestFailure) {
 384                         ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
 385                     } else {
 386                         ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
 387                     }
 388
 389                     recordSuccessfulRequest(req);
 390                     LOG.debug("Transaction {} canCommit completed", this);
 391                 });
 392
 393                 return;
 394             }
 395         }
 396
 397         // We have had some interference with successor injection, wait for it to complete and defer to the successor.
 398         awaitSuccessor().canCommit(ret);
 399     }
 400
 401     private AbstractProxyTransaction awaitSuccessor() {
 402         return getSuccessorState().await();
 403     }
 404
 405     final void preCommit(final VotingFuture<?> ret) {
 406         checkReadWrite();
 407         checkSealed();
 408
 409         final TransactionRequest<?> req = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
 410             localActor());
 411         sendRequest(req, t -> {
 412             if (t instanceof TransactionPreCommitSuccess) {
 413                 ret.voteYes();
 414             } else if (t instanceof RequestFailure) {
 415                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
 416             } else {
 417                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
 418             }
 419
 420             onPreCommitComplete(req);
 421         });
 422     }
 423
 424     private void onPreCommitComplete(final TransactionRequest<?> req) {
 425         /*
 426          * The backend has agreed that the transaction has entered PRE_COMMIT phase, meaning it will be committed
 427          * to storage after the timeout completes.
 428          *
 429          * All state has been replicated to the backend, hence we do not need to keep it around. Retain only
 430          * the precommit request, so we know which request to use for resync.
 431          */
 432         LOG.debug("Transaction {} preCommit completed, clearing successfulRequests", this);
 433         successfulRequests.clear();
 434
 435         // TODO: this works, but can contain some useless state (like batched operations). Create an empty
 436         //       equivalent of this request and store that.
 437         recordSuccessfulRequest(req);
 438     }
 439
 440     final void doCommit(final VotingFuture<?> ret) {
 441         checkReadWrite();
 442         checkSealed();
 443
 444         sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
 445             if (t instanceof TransactionCommitSuccess) {
 446                 ret.voteYes();
 447             } else if (t instanceof RequestFailure) {
 448                 ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
 449             } else {
 450                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
 451             }
 452
 453             LOG.debug("Transaction {} doCommit completed", this);
 454             purge();
 455         });
 456     }
 457
 458     void purge() {
 459         successfulRequests.clear();
 460
 461         final TransactionRequest<?> req = new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor());
 462         sendRequest(req, t -> {
 463             LOG.debug("Transaction {} purge completed", this);
 464             parent.completeTransaction(this);
 465         });
 466     }
 467
 468     // Called with the connection unlocked
 469     final synchronized void startReconnect() {
 470         // At this point canCommit/directCommit are blocked, we assert a new successor state, retrieving the previous
 471         // state. This method is called with the queue still unlocked.
 472         final SuccessorState nextState = new SuccessorState();
 473         final State prevState = STATE_UPDATER.getAndSet(this, nextState);
 474
 475         LOG.debug("Start reconnect of proxy {} previous state {}", this, prevState);
 476         Verify.verify(!(prevState instanceof SuccessorState), "Proxy %s duplicate reconnect attempt after %s", this,
 477             prevState);
 478
 479         // We have asserted a slow-path state, seal(), canCommit(), directCommit() are forced to slow paths, which will
 480         // wait until we unblock nextState's latch before accessing state. Now we record prevState for later use and we
 481         // are done.
 482         nextState.setPrevState(prevState);
 483     }
 484
 485     // Called with the connection locked
 486     final void replayMessages(final AbstractProxyTransaction successor,
 487             final Iterable<ConnectionEntry> enqueuedEntries) {
 488         final SuccessorState local = getSuccessorState();
 489         local.setSuccessor(successor);
 490
 491         // Replay successful requests first
 492         for (Object obj : successfulRequests) {
 493             if (obj instanceof TransactionRequest) {
 494                 LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
 495                 successor.replay((TransactionRequest<?>) obj, response -> { });
 496             } else {
 497                 Verify.verify(obj instanceof IncrementSequence);
 498                 successor.incrementSequence(((IncrementSequence) obj).getDelta());
 499             }
 500         }
 501         LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
 502         successfulRequests.clear();
 503
 504         // Now replay whatever is in the connection
 505         final Iterator<ConnectionEntry> it = enqueuedEntries.iterator();
 506         while (it.hasNext()) {
 507             final ConnectionEntry e = it.next();
 508             final Request<?, ?> req = e.getRequest();
 509
 510             if (getIdentifier().equals(req.getTarget())) {
 511                 Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
 512                 LOG.debug("Forwarding queued request {} to successor {}", req, successor);
 513                 successor.replay((TransactionRequest<?>) req, e.getCallback());
 514                 it.remove();
 515             }
 516         }
 517
 518         /*
 519          * Check the state at which we have started the reconnect attempt. State transitions triggered while we were
 520          * reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
 521          * at the end of this method.
 522          */
 523         final State prevState = local.getPrevState();
 524         if (SEALED.equals(prevState)) {
 525             LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
 526             flushState(successor);
 527             successor.ensureSealed();
 528         }
 529     }
 530
 531     /**
 532      * Invoked from {@link #replayMessages(AbstractProxyTransaction, Iterable)} to have successor adopt an in-flight
 533      * request.
 534      *
 535      * <p>
 536      * Note: this method is invoked by the predecessor on the successor.
 537      *
 538      * @param request Request which needs to be forwarded
 539      * @param callback Callback to be invoked once the request completes
 540      */
 541     private void replay(TransactionRequest<?> request, Consumer<Response<?, ?>> callback) {
 542         if (request instanceof AbstractLocalTransactionRequest) {
 543             handleForwardedLocalRequest((AbstractLocalTransactionRequest<?>) request, callback);
 544         } else {
 545             handleForwardedRemoteRequest(request, callback);
 546         }
 547     }
 548
 549     // Called with the connection locked
 550     final void finishReconnect() {
 551         final SuccessorState local = getSuccessorState();
 552         LOG.debug("Finishing reconnect of proxy {}", this);
 553
 554         // All done, release the latch, unblocking seal() and canCommit() slow paths
 555         local.finish();
 556     }
 557
 558     /**
 559      * Invoked from a retired connection for requests which have been in-flight and need to be re-adjusted
 560      * and forwarded to the successor connection.
 561      *
 562      * @param request Request to be forwarded
 563      * @param callback Original callback
 564      */
 565     final void forwardRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
 566         final AbstractProxyTransaction successor = getSuccessorState().getSuccessor();
 567
 568         if (successor instanceof LocalProxyTransaction) {
 569             forwardToLocal((LocalProxyTransaction)successor, request, callback);
 570         } else if (successor instanceof RemoteProxyTransaction) {
 571             forwardToRemote((RemoteProxyTransaction)successor, request, callback);
 572         } else {
 573             throw new IllegalStateException("Unhandled successor " + successor);
 574         }
 575     }
 576
 577     abstract boolean isSnapshotOnly();
 578
 579     abstract void doDelete(YangInstanceIdentifier path);
 580
 581     abstract void doMerge(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 582
 583     abstract void doWrite(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 584
 585     abstract CheckedFuture<Boolean, ReadFailedException> doExists(YangInstanceIdentifier path);
 586
 587     abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(YangInstanceIdentifier path);
 588
 589     abstract void doSeal();
 590
 591     abstract void doAbort();
 592
 593     @GuardedBy("this")
 594     abstract void flushState(AbstractProxyTransaction successor);
 595
 596     abstract TransactionRequest<?> commitRequest(boolean coordinated);
 597
 598     /**
 599      * Replay a request originating in this proxy to a successor remote proxy.
 600      */
 601     abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
 602             Consumer<Response<?, ?>> callback);
 603
 604     /**
 605      * Replay a request originating in this proxy to a successor local proxy.
 606      */
 607     abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
 608             Consumer<Response<?, ?>> callback);
 609
 610     /**
 611      * Invoked from {@link LocalProxyTransaction} when it replays its successful requests to its successor.
 612      *
 613      * <p>
 614      * Note: this method is invoked by the predecessor on the successor.
 615      *
 616      * @param request Request which needs to be forwarded
 617      * @param callback Callback to be invoked once the request completes
 618      */
 619     abstract void handleForwardedLocalRequest(AbstractLocalTransactionRequest<?> request,
 620             @Nullable Consumer<Response<?, ?>> callback);
 621
 622     /**
 623      * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor.
 624      *
 625      * <p>
 626      * Note: this method is invoked by the predecessor on the successor.
 627      *
 628      * @param request Request which needs to be forwarded
 629      * @param callback Callback to be invoked once the request completes
 630      */
 631     abstract void handleForwardedRemoteRequest(TransactionRequest<?> request,
 632             @Nullable Consumer<Response<?, ?>> callback);
 633
 634     @Override
 635     public final String toString() {
 636         return MoreObjects.toStringHelper(this).add("identifier", getIdentifier()).add("state", state).toString();
 637     }
 638 }