opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.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.datastore;
   9
  10 import static com.google.common.base.Preconditions.checkState;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.primitives.UnsignedLong;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import java.util.Collection;
  16 import java.util.Optional;
  17 import org.eclipse.jdt.annotation.Nullable;
  18 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
  19 import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
  20 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
  21 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess;
  22 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
  23 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionSuccess;
  24 import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
  25 import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
  26 import org.opendaylight.controller.cluster.access.commands.ReadTransactionSuccess;
  27 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
  28 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
  29 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
  30 import org.opendaylight.controller.cluster.access.commands.TransactionCommitSuccess;
  31 import org.opendaylight.controller.cluster.access.commands.TransactionDelete;
  32 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
  33 import org.opendaylight.controller.cluster.access.commands.TransactionMerge;
  34 import org.opendaylight.controller.cluster.access.commands.TransactionModification;
  35 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
  36 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
  37 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
  38 import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
  39 import org.opendaylight.controller.cluster.access.commands.TransactionWrite;
  40 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
  41 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  42 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
  43 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  44 import org.opendaylight.controller.cluster.access.concepts.UnsupportedRequestException;
  45 import org.opendaylight.yangtools.yang.common.Empty;
  46 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  47 import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate;
  48 import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51
  52 /**
  53  * Frontend read-write transaction state as observed by the shard leader. This class is NOT thread-safe.
  54  *
  55  * @author Robert Varga
  56  */
  57 final class FrontendReadWriteTransaction extends FrontendTransaction {
  58     private enum CommitStage {
  59         READY,
  60         CAN_COMMIT_PENDING,
  61         CAN_COMMIT_COMPLETE,
  62         PRE_COMMIT_PENDING,
  63         PRE_COMMIT_COMPLETE,
  64         COMMIT_PENDING,
  65     }
  66
  67     private abstract static class State {
  68         @Override
  69         public abstract String toString();
  70     }
  71
  72     private static final class Failed extends State {
  73         final RequestException cause;
  74
  75         Failed(final RequestException cause) {
  76             this.cause = requireNonNull(cause);
  77         }
  78
  79         @Override
  80         public String toString() {
  81             return "FAILED (" + cause.getMessage() + ")";
  82         }
  83     }
  84
  85     private static final class Open extends State {
  86         final ReadWriteShardDataTreeTransaction openTransaction;
  87
  88         Open(final ReadWriteShardDataTreeTransaction openTransaction) {
  89             this.openTransaction = requireNonNull(openTransaction);
  90         }
  91
  92         @Override
  93         public String toString() {
  94             return "OPEN";
  95         }
  96     }
  97
  98     private static final class Ready extends State {
  99         final ShardDataTreeCohort readyCohort;
 100         CommitStage stage;
 101
 102         Ready(final ShardDataTreeCohort readyCohort) {
 103             this.readyCohort = requireNonNull(readyCohort);
 104             stage = CommitStage.READY;
 105         }
 106
 107         @Override
 108         public String toString() {
 109             return "READY (" + stage + ")";
 110         }
 111     }
 112
 113     private static final class Sealed extends State {
 114         final DataTreeModification sealedModification;
 115
 116         Sealed(final DataTreeModification sealedModification) {
 117             this.sealedModification = requireNonNull(sealedModification);
 118         }
 119
 120         @Override
 121         public String toString() {
 122             return "SEALED";
 123         }
 124     }
 125
 126     /**
 127      * Retired state, needed to catch and suppress callbacks after we have removed associated state.
 128      */
 129     private static final class Retired extends State {
 130         private final String prevStateString;
 131
 132         Retired(final State prevState) {
 133             prevStateString = prevState.toString();
 134         }
 135
 136         @Override
 137         public String toString() {
 138             return "RETIRED (in " + prevStateString + ")";
 139         }
 140     }
 141
 142     private static final Logger LOG = LoggerFactory.getLogger(FrontendReadWriteTransaction.class);
 143     private static final State ABORTED = new State() {
 144         @Override
 145         public String toString() {
 146             return "ABORTED";
 147         }
 148     };
 149     private static final State ABORTING = new State() {
 150         @Override
 151         public String toString() {
 152             return "ABORTING";
 153         }
 154     };
 155     private static final State COMMITTED = new State() {
 156         @Override
 157         public String toString() {
 158             return "COMMITTED";
 159         }
 160     };
 161
 162     private State state;
 163
 164     private FrontendReadWriteTransaction(final AbstractFrontendHistory history, final TransactionIdentifier id,
 165             final ReadWriteShardDataTreeTransaction transaction) {
 166         super(history, id);
 167         state = new Open(transaction);
 168     }
 169
 170     private FrontendReadWriteTransaction(final AbstractFrontendHistory history, final TransactionIdentifier id,
 171             final DataTreeModification mod) {
 172         super(history, id);
 173         state = new Sealed(mod);
 174     }
 175
 176     static FrontendReadWriteTransaction createOpen(final AbstractFrontendHistory history,
 177             final ReadWriteShardDataTreeTransaction transaction) {
 178         return new FrontendReadWriteTransaction(history, transaction.getIdentifier(), transaction);
 179     }
 180
 181     static FrontendReadWriteTransaction createReady(final AbstractFrontendHistory history,
 182             final TransactionIdentifier id, final DataTreeModification mod) {
 183         return new FrontendReadWriteTransaction(history, id, mod);
 184     }
 185
 186     // Sequence has already been checked
 187     @Override
 188     TransactionSuccess<?> doHandleRequest(final TransactionRequest<?> request, final RequestEnvelope envelope,
 189             final long now) throws RequestException {
 190         if (request instanceof ModifyTransactionRequest) {
 191             return handleModifyTransaction((ModifyTransactionRequest) request, envelope, now);
 192         } else if (request instanceof CommitLocalTransactionRequest) {
 193             handleCommitLocalTransaction((CommitLocalTransactionRequest) request, envelope, now);
 194             return null;
 195         } else if (request instanceof ExistsTransactionRequest) {
 196             return handleExistsTransaction((ExistsTransactionRequest) request);
 197         } else if (request instanceof ReadTransactionRequest) {
 198             return handleReadTransaction((ReadTransactionRequest) request);
 199         } else if (request instanceof TransactionPreCommitRequest) {
 200             handleTransactionPreCommit((TransactionPreCommitRequest) request, envelope, now);
 201             return null;
 202         } else if (request instanceof TransactionDoCommitRequest) {
 203             handleTransactionDoCommit((TransactionDoCommitRequest) request, envelope, now);
 204             return null;
 205         } else if (request instanceof TransactionAbortRequest) {
 206             return handleTransactionAbort(request.getSequence(), envelope, now);
 207         } else if (request instanceof AbortLocalTransactionRequest) {
 208             handleLocalTransactionAbort(request.getSequence(), envelope, now);
 209             return null;
 210         } else {
 211             LOG.warn("Rejecting unsupported request {}", request);
 212             throw new UnsupportedRequestException(request);
 213         }
 214     }
 215
 216     @Override
 217     void retire() {
 218         state = new Retired(state);
 219     }
 220
 221     private void handleTransactionPreCommit(final TransactionPreCommitRequest request,
 222             final RequestEnvelope envelope, final long now) throws RequestException {
 223         throwIfFailed();
 224
 225         final Ready ready = checkReady();
 226         switch (ready.stage) {
 227             case PRE_COMMIT_PENDING:
 228                 LOG.debug("{}: Transaction {} is already preCommitting", persistenceId(), getIdentifier());
 229                 break;
 230             case CAN_COMMIT_COMPLETE:
 231                 ready.stage = CommitStage.PRE_COMMIT_PENDING;
 232                 LOG.debug("{}: Transaction {} initiating preCommit", persistenceId(), getIdentifier());
 233                 ready.readyCohort.preCommit(new FutureCallback<DataTreeCandidate>() {
 234                     @Override
 235                     public void onSuccess(final DataTreeCandidate result) {
 236                         successfulPreCommit(envelope, now);
 237                     }
 238
 239                     @Override
 240                     public void onFailure(final Throwable failure) {
 241                         failTransaction(envelope, now, new RuntimeRequestException("Precommit failed", failure));
 242                     }
 243                 });
 244                 break;
 245             case CAN_COMMIT_PENDING:
 246             case COMMIT_PENDING:
 247             case PRE_COMMIT_COMPLETE:
 248             case READY:
 249                 throw new IllegalStateException("Attempted to preCommit in stage " + ready.stage);
 250             default:
 251                 throwUnhandledCommitStage(ready);
 252         }
 253     }
 254
 255     void successfulPreCommit(final RequestEnvelope envelope, final long startTime) {
 256         if (state instanceof Retired) {
 257             LOG.debug("{}: Suppressing successful preCommit of retired transaction {}", persistenceId(),
 258                 getIdentifier());
 259             return;
 260         }
 261
 262         final Ready ready = checkReady();
 263         LOG.debug("{}: Transaction {} completed preCommit", persistenceId(), getIdentifier());
 264         recordAndSendSuccess(envelope, startTime, new TransactionPreCommitSuccess(getIdentifier(),
 265             envelope.getMessage().getSequence()));
 266         ready.stage = CommitStage.PRE_COMMIT_COMPLETE;
 267     }
 268
 269     void failTransaction(final RequestEnvelope envelope, final long now, final RuntimeRequestException cause) {
 270         if (state instanceof Retired) {
 271             LOG.debug("{}: Suppressing failure of retired transaction {}", persistenceId(), getIdentifier(), cause);
 272             return;
 273         }
 274
 275         recordAndSendFailure(envelope, now, cause);
 276         state = new Failed(cause);
 277         LOG.debug("{}: Transaction {} failed", persistenceId(), getIdentifier(), cause);
 278     }
 279
 280     private void handleTransactionDoCommit(final TransactionDoCommitRequest request, final RequestEnvelope envelope,
 281             final long now) throws RequestException {
 282         throwIfFailed();
 283
 284         final Ready ready = checkReady();
 285         switch (ready.stage) {
 286             case COMMIT_PENDING:
 287                 LOG.debug("{}: Transaction {} is already committing", persistenceId(), getIdentifier());
 288                 break;
 289             case PRE_COMMIT_COMPLETE:
 290                 ready.stage = CommitStage.COMMIT_PENDING;
 291                 LOG.debug("{}: Transaction {} initiating commit", persistenceId(), getIdentifier());
 292                 ready.readyCohort.commit(new FutureCallback<UnsignedLong>() {
 293                     @Override
 294                     public void onSuccess(final UnsignedLong result) {
 295                         successfulCommit(envelope, now);
 296                     }
 297
 298                     @Override
 299                     public void onFailure(final Throwable failure) {
 300                         failTransaction(envelope, now, new RuntimeRequestException("Commit failed", failure));
 301                     }
 302                 });
 303                 break;
 304             case CAN_COMMIT_COMPLETE:
 305             case CAN_COMMIT_PENDING:
 306             case PRE_COMMIT_PENDING:
 307             case READY:
 308                 throw new IllegalStateException("Attempted to doCommit in stage " + ready.stage);
 309             default:
 310                 throwUnhandledCommitStage(ready);
 311         }
 312     }
 313
 314     private void handleLocalTransactionAbort(final long sequence, final RequestEnvelope envelope, final long now) {
 315         checkOpen().abort(() -> recordAndSendSuccess(envelope, now, new TransactionAbortSuccess(getIdentifier(),
 316             sequence)));
 317     }
 318
 319     private void startAbort() {
 320         state = ABORTING;
 321         LOG.debug("{}: Transaction {} aborting", persistenceId(), getIdentifier());
 322     }
 323
 324     private void finishAbort() {
 325         state = ABORTED;
 326         LOG.debug("{}: Transaction {} aborted", persistenceId(), getIdentifier());
 327     }
 328
 329     private TransactionAbortSuccess handleTransactionAbort(final long sequence, final RequestEnvelope envelope,
 330             final long now) {
 331         if (state instanceof Open) {
 332             final ReadWriteShardDataTreeTransaction openTransaction = checkOpen();
 333             startAbort();
 334             openTransaction.abort(() -> {
 335                 recordAndSendSuccess(envelope, now, new TransactionAbortSuccess(getIdentifier(),
 336                     sequence));
 337                 finishAbort();
 338             });
 339             return null;
 340         }
 341         if (ABORTING.equals(state)) {
 342             LOG.debug("{}: Transaction {} already aborting", persistenceId(), getIdentifier());
 343             return null;
 344         }
 345         if (ABORTED.equals(state)) {
 346             // We should have recorded the reply
 347             LOG.warn("{}: Transaction {} already aborted", persistenceId(), getIdentifier());
 348             return new TransactionAbortSuccess(getIdentifier(), sequence);
 349         }
 350
 351         final Ready ready = checkReady();
 352         startAbort();
 353         ready.readyCohort.abort(new FutureCallback<>() {
 354             @Override
 355             public void onSuccess(final Empty result) {
 356                 recordAndSendSuccess(envelope, now, new TransactionAbortSuccess(getIdentifier(), sequence));
 357                 finishAbort();
 358             }
 359
 360             @Override
 361             public void onFailure(final Throwable failure) {
 362                 recordAndSendFailure(envelope, now, new RuntimeRequestException("Abort failed", failure));
 363                 LOG.warn("{}: Transaction {} abort failed", persistenceId(), getIdentifier(), failure);
 364                 finishAbort();
 365             }
 366         });
 367         return null;
 368     }
 369
 370     private void coordinatedCommit(final RequestEnvelope envelope, final long now) throws RequestException {
 371         throwIfFailed();
 372
 373         final Ready ready = checkReady();
 374         switch (ready.stage) {
 375             case CAN_COMMIT_PENDING:
 376                 LOG.debug("{}: Transaction {} is already canCommitting", persistenceId(), getIdentifier());
 377                 break;
 378             case READY:
 379                 ready.stage = CommitStage.CAN_COMMIT_PENDING;
 380                 LOG.debug("{}: Transaction {} initiating canCommit", persistenceId(), getIdentifier());
 381                 checkReady().readyCohort.canCommit(new FutureCallback<>() {
 382                     @Override
 383                     public void onSuccess(final Empty result) {
 384                         successfulCanCommit(envelope, now);
 385                     }
 386
 387                     @Override
 388                     public void onFailure(final Throwable failure) {
 389                         failTransaction(envelope, now, new RuntimeRequestException("CanCommit failed", failure));
 390                     }
 391                 });
 392                 break;
 393             case CAN_COMMIT_COMPLETE:
 394             case COMMIT_PENDING:
 395             case PRE_COMMIT_COMPLETE:
 396             case PRE_COMMIT_PENDING:
 397                 throw new IllegalStateException("Attempted to canCommit in stage " + ready.stage);
 398             default:
 399                 throwUnhandledCommitStage(ready);
 400         }
 401     }
 402
 403     void successfulCanCommit(final RequestEnvelope envelope, final long startTime) {
 404         if (state instanceof Retired) {
 405             LOG.debug("{}: Suppressing successful canCommit of retired transaction {}", persistenceId(),
 406                 getIdentifier());
 407             return;
 408         }
 409
 410         final Ready ready = checkReady();
 411         recordAndSendSuccess(envelope, startTime, new TransactionCanCommitSuccess(getIdentifier(),
 412             envelope.getMessage().getSequence()));
 413         ready.stage = CommitStage.CAN_COMMIT_COMPLETE;
 414         LOG.debug("{}: Transaction {} completed canCommit", persistenceId(), getIdentifier());
 415     }
 416
 417     private void directCommit(final RequestEnvelope envelope, final long now) throws RequestException {
 418         throwIfFailed();
 419
 420         final Ready ready = checkReady();
 421         switch (ready.stage) {
 422             case CAN_COMMIT_COMPLETE:
 423             case CAN_COMMIT_PENDING:
 424             case COMMIT_PENDING:
 425             case PRE_COMMIT_COMPLETE:
 426             case PRE_COMMIT_PENDING:
 427                 LOG.debug("{}: Transaction {} in state {}, not initiating direct commit for {}", persistenceId(),
 428                     getIdentifier(), state, envelope);
 429                 break;
 430             case READY:
 431                 ready.stage = CommitStage.CAN_COMMIT_PENDING;
 432                 LOG.debug("{}: Transaction {} initiating direct canCommit", persistenceId(), getIdentifier());
 433                 ready.readyCohort.canCommit(new FutureCallback<>() {
 434                     @Override
 435                     public void onSuccess(final Empty result) {
 436                         successfulDirectCanCommit(envelope, now);
 437                     }
 438
 439                     @Override
 440                     public void onFailure(final Throwable failure) {
 441                         failTransaction(envelope, now, new RuntimeRequestException("CanCommit failed", failure));
 442                     }
 443                 });
 444                 break;
 445             default:
 446                 throwUnhandledCommitStage(ready);
 447         }
 448     }
 449
 450     void successfulDirectCanCommit(final RequestEnvelope envelope, final long startTime) {
 451         if (state instanceof Retired) {
 452             LOG.debug("{}: Suppressing direct canCommit of retired transaction {}", persistenceId(), getIdentifier());
 453             return;
 454         }
 455
 456         final Ready ready = checkReady();
 457         ready.stage = CommitStage.PRE_COMMIT_PENDING;
 458         LOG.debug("{}: Transaction {} initiating direct preCommit", persistenceId(), getIdentifier());
 459         ready.readyCohort.preCommit(new FutureCallback<DataTreeCandidate>() {
 460             @Override
 461             public void onSuccess(final DataTreeCandidate result) {
 462                 successfulDirectPreCommit(envelope, startTime);
 463             }
 464
 465             @Override
 466             public void onFailure(final Throwable failure) {
 467                 failTransaction(envelope, startTime, new RuntimeRequestException("PreCommit failed", failure));
 468             }
 469         });
 470     }
 471
 472     void successfulDirectPreCommit(final RequestEnvelope envelope, final long startTime) {
 473         if (state instanceof Retired) {
 474             LOG.debug("{}: Suppressing direct commit of retired transaction {}", persistenceId(), getIdentifier());
 475             return;
 476         }
 477
 478         final Ready ready = checkReady();
 479         ready.stage = CommitStage.COMMIT_PENDING;
 480         LOG.debug("{}: Transaction {} initiating direct commit", persistenceId(), getIdentifier());
 481         ready.readyCohort.commit(new FutureCallback<UnsignedLong>() {
 482             @Override
 483             public void onSuccess(final UnsignedLong result) {
 484                 successfulCommit(envelope, startTime);
 485             }
 486
 487             @Override
 488             public void onFailure(final Throwable failure) {
 489                 failTransaction(envelope, startTime, new RuntimeRequestException("DoCommit failed", failure));
 490             }
 491         });
 492     }
 493
 494     void successfulCommit(final RequestEnvelope envelope, final long startTime) {
 495         if (state instanceof Retired) {
 496             LOG.debug("{}: Suppressing commit response on retired transaction {}", persistenceId(), getIdentifier());
 497             return;
 498         }
 499
 500         recordAndSendSuccess(envelope, startTime, new TransactionCommitSuccess(getIdentifier(),
 501             envelope.getMessage().getSequence()));
 502         state = COMMITTED;
 503     }
 504
 505     private void handleCommitLocalTransaction(final CommitLocalTransactionRequest request,
 506             final RequestEnvelope envelope, final long now) throws RequestException {
 507         final DataTreeModification sealedModification = checkSealed();
 508         if (!sealedModification.equals(request.getModification())) {
 509             LOG.warn("Expecting modification {}, commit request has {}", sealedModification, request.getModification());
 510             throw new UnsupportedRequestException(request);
 511         }
 512
 513         final Optional<Exception> optFailure = request.getDelayedFailure();
 514         if (optFailure.isPresent()) {
 515             state = new Ready(history().createFailedCohort(getIdentifier(), sealedModification, optFailure.get()));
 516         } else {
 517             state = new Ready(history().createReadyCohort(getIdentifier(), sealedModification, Optional.empty()));
 518         }
 519
 520         if (request.isCoordinated()) {
 521             coordinatedCommit(envelope, now);
 522         } else {
 523             directCommit(envelope, now);
 524         }
 525     }
 526
 527     private ExistsTransactionSuccess handleExistsTransaction(final ExistsTransactionRequest request) {
 528         final Optional<NormalizedNode> data = checkOpen().getSnapshot().readNode(request.getPath());
 529         return recordSuccess(request.getSequence(), new ExistsTransactionSuccess(getIdentifier(), request.getSequence(),
 530             data.isPresent()));
 531     }
 532
 533     private ReadTransactionSuccess handleReadTransaction(final ReadTransactionRequest request) {
 534         final Optional<NormalizedNode> data = checkOpen().getSnapshot().readNode(request.getPath());
 535         return recordSuccess(request.getSequence(), new ReadTransactionSuccess(getIdentifier(), request.getSequence(),
 536             data));
 537     }
 538
 539     private ModifyTransactionSuccess replyModifySuccess(final long sequence) {
 540         return recordSuccess(sequence, new ModifyTransactionSuccess(getIdentifier(), sequence));
 541     }
 542
 543     private void applyModifications(final Collection<TransactionModification> modifications) {
 544         if (!modifications.isEmpty()) {
 545             final DataTreeModification modification = checkOpen().getSnapshot();
 546             for (TransactionModification m : modifications) {
 547                 if (m instanceof TransactionDelete) {
 548                     modification.delete(m.getPath());
 549                 } else if (m instanceof TransactionWrite) {
 550                     modification.write(m.getPath(), ((TransactionWrite) m).getData());
 551                 } else if (m instanceof TransactionMerge) {
 552                     modification.merge(m.getPath(), ((TransactionMerge) m).getData());
 553                 } else {
 554                     LOG.warn("{}: ignoring unhandled modification {}", persistenceId(), m);
 555                 }
 556             }
 557         }
 558     }
 559
 560     private @Nullable TransactionSuccess<?> handleModifyTransaction(final ModifyTransactionRequest request,
 561             final RequestEnvelope envelope, final long now) throws RequestException {
 562         // We need to examine the persistence protocol first to see if this is an idempotent request. If there is no
 563         // protocol, there is nothing for us to do.
 564         final Optional<PersistenceProtocol> maybeProto = request.getPersistenceProtocol();
 565         if (!maybeProto.isPresent()) {
 566             applyModifications(request.getModifications());
 567             return replyModifySuccess(request.getSequence());
 568         }
 569
 570         switch (maybeProto.get()) {
 571             case ABORT:
 572                 if (ABORTING.equals(state)) {
 573                     LOG.debug("{}: Transaction {} already aborting", persistenceId(), getIdentifier());
 574                     return null;
 575                 }
 576                 final ReadWriteShardDataTreeTransaction openTransaction = checkOpen();
 577                 startAbort();
 578                 openTransaction.abort(() -> {
 579                     recordAndSendSuccess(envelope, now, new ModifyTransactionSuccess(getIdentifier(),
 580                         request.getSequence()));
 581                     finishAbort();
 582                 });
 583                 return null;
 584             case READY:
 585                 ensureReady(request.getModifications());
 586                 return replyModifySuccess(request.getSequence());
 587             case SIMPLE:
 588                 ensureReady(request.getModifications());
 589                 directCommit(envelope, now);
 590                 return null;
 591             case THREE_PHASE:
 592                 ensureReady(request.getModifications());
 593                 coordinatedCommit(envelope, now);
 594                 return null;
 595             default:
 596                 LOG.warn("{}: rejecting unsupported protocol {}", persistenceId(), maybeProto.get());
 597                 throw new UnsupportedRequestException(request);
 598         }
 599     }
 600
 601     private void ensureReady(final Collection<TransactionModification> modifications) {
 602         // We may have a combination of READY + SIMPLE/THREE_PHASE , in which case we want to ready the transaction
 603         // only once.
 604         if (state instanceof Ready) {
 605             LOG.debug("{}: {} is already in state {}", persistenceId(), getIdentifier(), state);
 606             return;
 607         }
 608
 609         applyModifications(modifications);
 610         state = new Ready(checkOpen().ready(Optional.empty()));
 611         LOG.debug("{}: transitioned {} to ready", persistenceId(), getIdentifier());
 612     }
 613
 614     private void throwIfFailed() throws RequestException {
 615         if (state instanceof Failed) {
 616             LOG.debug("{}: {} has failed, rejecting request", persistenceId(), getIdentifier());
 617             throw ((Failed) state).cause;
 618         }
 619     }
 620
 621     private ReadWriteShardDataTreeTransaction checkOpen() {
 622         checkState(state instanceof Open, "%s expect to be open, is in state %s", getIdentifier(), state);
 623         return ((Open) state).openTransaction;
 624     }
 625
 626     private Ready checkReady() {
 627         checkState(state instanceof Ready, "%s expect to be ready, is in state %s", getIdentifier(), state);
 628         return (Ready) state;
 629     }
 630
 631     private DataTreeModification checkSealed() {
 632         checkState(state instanceof Sealed, "%s expect to be sealed, is in state %s", getIdentifier(), state);
 633         return ((Sealed) state).sealedModification;
 634     }
 635
 636     private static void throwUnhandledCommitStage(final Ready ready) {
 637         throw new IllegalStateException("Unhandled commit stage " + ready.stage);
 638     }
 639 }