X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FFrontendReadWriteTransaction.java;h=4bd2a5725758b228c1803414b74e116810ca40b9;hb=HEAD;hp=2b444a6b579901bde4c10e0531ae4092218cdb4c;hpb=8232a626b43fdd2f5799da0fbcfb0f02d3c8f4fb;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java index 2b444a6b57..c626791547 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java @@ -7,13 +7,14 @@ */ package org.opendaylight.controller.cluster.datastore; -import com.google.common.base.Optional; -import com.google.common.base.Preconditions; +import static com.google.common.base.Preconditions.checkState; +import static java.util.Objects.requireNonNull; + import com.google.common.primitives.UnsignedLong; import com.google.common.util.concurrent.FutureCallback; import java.util.Collection; -import javax.annotation.Nullable; -import javax.annotation.concurrent.NotThreadSafe; +import java.util.Optional; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest; import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest; import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest; @@ -41,18 +42,18 @@ import org.opendaylight.controller.cluster.access.concepts.RequestException; import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.access.concepts.UnsupportedRequestException; +import org.opendaylight.yangtools.yang.common.Empty; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Frontend read-write transaction state as observed by the shard leader. + * Frontend read-write transaction state as observed by the shard leader. This class is NOT thread-safe. * * @author Robert Varga */ -@NotThreadSafe final class FrontendReadWriteTransaction extends FrontendTransaction { private enum CommitStage { READY, @@ -72,7 +73,7 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { final RequestException cause; Failed(final RequestException cause) { - this.cause = Preconditions.checkNotNull(cause); + this.cause = requireNonNull(cause); } @Override @@ -85,7 +86,7 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { final ReadWriteShardDataTreeTransaction openTransaction; Open(final ReadWriteShardDataTreeTransaction openTransaction) { - this.openTransaction = Preconditions.checkNotNull(openTransaction); + this.openTransaction = requireNonNull(openTransaction); } @Override @@ -99,8 +100,8 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { CommitStage stage; Ready(final ShardDataTreeCohort readyCohort) { - this.readyCohort = Preconditions.checkNotNull(readyCohort); - this.stage = CommitStage.READY; + this.readyCohort = requireNonNull(readyCohort); + stage = CommitStage.READY; } @Override @@ -113,7 +114,7 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { final DataTreeModification sealedModification; Sealed(final DataTreeModification sealedModification) { - this.sealedModification = Preconditions.checkNotNull(sealedModification); + this.sealedModification = requireNonNull(sealedModification); } @Override @@ -163,13 +164,13 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { private FrontendReadWriteTransaction(final AbstractFrontendHistory history, final TransactionIdentifier id, final ReadWriteShardDataTreeTransaction transaction) { super(history, id); - this.state = new Open(transaction); + state = new Open(transaction); } private FrontendReadWriteTransaction(final AbstractFrontendHistory history, final TransactionIdentifier id, final DataTreeModification mod) { super(history, id); - this.state = new Sealed(mod); + state = new Sealed(mod); } static FrontendReadWriteTransaction createOpen(final AbstractFrontendHistory history, @@ -184,22 +185,22 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { // Sequence has already been checked @Override - @Nullable TransactionSuccess doHandleRequest(final TransactionRequest request, final RequestEnvelope envelope, + TransactionSuccess doHandleRequest(final TransactionRequest request, final RequestEnvelope envelope, final long now) throws RequestException { - if (request instanceof ModifyTransactionRequest) { - return handleModifyTransaction((ModifyTransactionRequest) request, envelope, now); - } else if (request instanceof CommitLocalTransactionRequest) { - handleCommitLocalTransaction((CommitLocalTransactionRequest) request, envelope, now); + if (request instanceof ModifyTransactionRequest modifyRequest) { + return handleModifyTransaction(modifyRequest, envelope, now); + } else if (request instanceof CommitLocalTransactionRequest commitLocalRequest) { + handleCommitLocalTransaction(commitLocalRequest, envelope, now); return null; - } else if (request instanceof ExistsTransactionRequest) { - return handleExistsTransaction((ExistsTransactionRequest) request); - } else if (request instanceof ReadTransactionRequest) { - return handleReadTransaction((ReadTransactionRequest) request); - } else if (request instanceof TransactionPreCommitRequest) { - handleTransactionPreCommit((TransactionPreCommitRequest) request, envelope, now); + } else if (request instanceof ExistsTransactionRequest existsRequest) { + return handleExistsTransaction(existsRequest); + } else if (request instanceof ReadTransactionRequest readRequest) { + return handleReadTransaction(readRequest); + } else if (request instanceof TransactionPreCommitRequest preCommitRequest) { + handleTransactionPreCommit(preCommitRequest, envelope, now); return null; - } else if (request instanceof TransactionDoCommitRequest) { - handleTransactionDoCommit((TransactionDoCommitRequest) request, envelope, now); + } else if (request instanceof TransactionDoCommitRequest doCommitRequest) { + handleTransactionDoCommit(doCommitRequest, envelope, now); return null; } else if (request instanceof TransactionAbortRequest) { return handleTransactionAbort(request.getSequence(), envelope, now); @@ -349,9 +350,9 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { final Ready ready = checkReady(); startAbort(); - ready.readyCohort.abort(new FutureCallback() { + ready.readyCohort.abort(new FutureCallback<>() { @Override - public void onSuccess(final Void result) { + public void onSuccess(final Empty result) { recordAndSendSuccess(envelope, now, new TransactionAbortSuccess(getIdentifier(), sequence)); finishAbort(); } @@ -377,9 +378,9 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { case READY: ready.stage = CommitStage.CAN_COMMIT_PENDING; LOG.debug("{}: Transaction {} initiating canCommit", persistenceId(), getIdentifier()); - checkReady().readyCohort.canCommit(new FutureCallback() { + checkReady().readyCohort.canCommit(new FutureCallback<>() { @Override - public void onSuccess(final Void result) { + public void onSuccess(final Empty result) { successfulCanCommit(envelope, now); } @@ -429,9 +430,9 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { case READY: ready.stage = CommitStage.CAN_COMMIT_PENDING; LOG.debug("{}: Transaction {} initiating direct canCommit", persistenceId(), getIdentifier()); - ready.readyCohort.canCommit(new FutureCallback() { + ready.readyCohort.canCommit(new FutureCallback<>() { @Override - public void onSuccess(final Void result) { + public void onSuccess(final Empty result) { successfulDirectCanCommit(envelope, now); } @@ -509,11 +510,12 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { throw new UnsupportedRequestException(request); } - final java.util.Optional optFailure = request.getDelayedFailure(); + final Optional optFailure = request.getDelayedFailure(); if (optFailure.isPresent()) { - state = new Ready(history().createFailedCohort(getIdentifier(), sealedModification, optFailure.get())); + state = new Ready(history().createFailedCohort(getIdentifier(), sealedModification, + optFailure.orElseThrow())); } else { - state = new Ready(history().createReadyCohort(getIdentifier(), sealedModification)); + state = new Ready(history().createReadyCohort(getIdentifier(), sealedModification, Optional.empty())); } if (request.isCoordinated()) { @@ -523,16 +525,14 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { } } - private ExistsTransactionSuccess handleExistsTransaction(final ExistsTransactionRequest request) - throws RequestException { - final Optional> data = checkOpen().getSnapshot().readNode(request.getPath()); + private ExistsTransactionSuccess handleExistsTransaction(final ExistsTransactionRequest request) { + final Optional data = checkOpen().getSnapshot().readNode(request.getPath()); return recordSuccess(request.getSequence(), new ExistsTransactionSuccess(getIdentifier(), request.getSequence(), data.isPresent())); } - private ReadTransactionSuccess handleReadTransaction(final ReadTransactionRequest request) - throws RequestException { - final Optional> data = checkOpen().getSnapshot().readNode(request.getPath()); + private ReadTransactionSuccess handleReadTransaction(final ReadTransactionRequest request) { + final Optional data = checkOpen().getSnapshot().readNode(request.getPath()); return recordSuccess(request.getSequence(), new ReadTransactionSuccess(getIdentifier(), request.getSequence(), data)); } @@ -547,10 +547,10 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { for (TransactionModification m : modifications) { if (m instanceof TransactionDelete) { modification.delete(m.getPath()); - } else if (m instanceof TransactionWrite) { - modification.write(m.getPath(), ((TransactionWrite) m).getData()); - } else if (m instanceof TransactionMerge) { - modification.merge(m.getPath(), ((TransactionMerge) m).getData()); + } else if (m instanceof TransactionWrite write) { + modification.write(m.getPath(), write.getData()); + } else if (m instanceof TransactionMerge merge) { + modification.merge(m.getPath(), merge.getData()); } else { LOG.warn("{}: ignoring unhandled modification {}", persistenceId(), m); } @@ -558,18 +558,17 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { } } - @Nullable - private TransactionSuccess handleModifyTransaction(final ModifyTransactionRequest request, + private @Nullable TransactionSuccess handleModifyTransaction(final ModifyTransactionRequest request, final RequestEnvelope envelope, final long now) throws RequestException { // We need to examine the persistence protocol first to see if this is an idempotent request. If there is no // protocol, there is nothing for us to do. - final java.util.Optional maybeProto = request.getPersistenceProtocol(); + final Optional maybeProto = request.getPersistenceProtocol(); if (!maybeProto.isPresent()) { applyModifications(request.getModifications()); return replyModifySuccess(request.getSequence()); } - switch (maybeProto.get()) { + switch (maybeProto.orElseThrow()) { case ABORT: if (ABORTING.equals(state)) { LOG.debug("{}: Transaction {} already aborting", persistenceId(), getIdentifier()); @@ -595,7 +594,7 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { coordinatedCommit(envelope, now); return null; default: - LOG.warn("{}: rejecting unsupported protocol {}", persistenceId(), maybeProto.get()); + LOG.warn("{}: rejecting unsupported protocol {}", persistenceId(), maybeProto.orElseThrow()); throw new UnsupportedRequestException(request); } } @@ -609,7 +608,7 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { } applyModifications(modifications); - state = new Ready(checkOpen().ready()); + state = new Ready(checkOpen().ready(Optional.empty())); LOG.debug("{}: transitioned {} to ready", persistenceId(), getIdentifier()); } @@ -621,20 +620,17 @@ final class FrontendReadWriteTransaction extends FrontendTransaction { } private ReadWriteShardDataTreeTransaction checkOpen() { - Preconditions.checkState(state instanceof Open, "%s expect to be open, is in state %s", getIdentifier(), - state); + checkState(state instanceof Open, "%s expect to be open, is in state %s", getIdentifier(), state); return ((Open) state).openTransaction; } private Ready checkReady() { - Preconditions.checkState(state instanceof Ready, "%s expect to be ready, is in state %s", getIdentifier(), - state); + checkState(state instanceof Ready, "%s expect to be ready, is in state %s", getIdentifier(), state); return (Ready) state; } private DataTreeModification checkSealed() { - Preconditions.checkState(state instanceof Sealed, "%s expect to be sealed, is in state %s", getIdentifier(), - state); + checkState(state instanceof Sealed, "%s expect to be sealed, is in state %s", getIdentifier(), state); return ((Sealed) state).sealedModification; }