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