X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=897527764c18b2ae26061b014ac22e5e24ab4de5;hb=bc301a153a07be7c23327e85179974422624c80a;hp=47a25d7ea339af7f9dd87933879148f3c06b8f6e;hpb=402dbc040ddb5dfc488320356b5a36c66d59c36e;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java index 47a25d7ea3..897527764c 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java @@ -5,7 +5,6 @@ * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ - package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorRef; @@ -31,8 +30,8 @@ import java.util.Collection; import java.util.Collections; import java.util.Map; import java.util.concurrent.TimeUnit; -import javax.annotation.Nonnull; -import javax.annotation.Nullable; +import org.eclipse.jdt.annotation.NonNull; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.controller.cluster.access.ABIVersion; import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest; import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess; @@ -107,7 +106,6 @@ import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailed import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.opendaylight.yangtools.yang.model.api.SchemaContextProvider; -import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; /** @@ -161,6 +159,8 @@ public class Shard extends RaftActor { /// The name of this shard private final String name; + private final String shardName; + private final ShardStats shardMBean; private final ShardDataTreeListenerInfoMXBeanImpl listenerInfoMXBean; @@ -201,6 +201,7 @@ public class Shard extends RaftActor { Optional.of(builder.getDatastoreContext().getShardRaftConfig()), DataStoreVersions.CURRENT_VERSION); this.name = builder.getId().toString(); + this.shardName = builder.getId().getShardName(); this.datastoreContext = builder.getDatastoreContext(); this.restoreFromSnapshot = builder.getRestoreFromSnapshot(); this.frontendMetadata = new FrontendMetadata(name); @@ -362,7 +363,9 @@ public class Shard extends RaftActor { (DataTreeCohortActorRegistry.CohortRegistryCommand) message); } else if (message instanceof PersistAbortTransactionPayload) { final TransactionIdentifier txId = ((PersistAbortTransactionPayload) message).getTransactionId(); - persistPayload(txId, AbortTransactionPayload.create(txId), true); + persistPayload(txId, AbortTransactionPayload.create( + txId, datastoreContext.getInitialPayloadSerializedBufferCapacity()), true); + store.purgeTransaction(txId, null); } else if (message instanceof MakeLeaderLocal) { onMakeLeaderLocal(); } else if (RESUME_NEXT_PENDING_TRANSACTION.equals(message)) { @@ -392,9 +395,7 @@ public class Shard extends RaftActor { responseMessageSlicer.slice(SliceOptions.builder().identifier(success.getTarget()) .message(envelope.newSuccessEnvelope(success, executionTimeNanos)) .sendTo(envelope.getMessage().getReplyTo()).replyTo(self()) - .onFailureCallback(t -> { - LOG.warn("Error slicing response {}", success, t); - }).build())); + .onFailureCallback(t -> LOG.warn("Error slicing response {}", success, t)).build())); } else { envelope.sendSuccess(success, executionTimeNanos); } @@ -460,8 +461,7 @@ public class Shard extends RaftActor { } // Acquire our frontend tracking handle and verify generation matches - @Nullable - private LeaderFrontendState findFrontend(final ClientIdentifier clientId) throws RequestException { + private @Nullable LeaderFrontendState findFrontend(final ClientIdentifier clientId) throws RequestException { final LeaderFrontendState existing = knownFrontends.get(clientId.getFrontendId()); if (existing != null) { final int cmp = Long.compareUnsigned(existing.getIdentifier().getGeneration(), clientId.getGeneration()); @@ -471,7 +471,8 @@ public class Shard extends RaftActor { } if (cmp > 0) { LOG.debug("{}: rejecting request from outdated client {}", persistenceId(), clientId); - throw new RetiredGenerationException(existing.getIdentifier().getGeneration()); + throw new RetiredGenerationException(clientId.getGeneration(), + existing.getIdentifier().getGeneration()); } LOG.info("{}: retiring state {}, outdated by request from client {}", persistenceId(), existing, clientId); @@ -494,8 +495,7 @@ public class Shard extends RaftActor { throw new OutOfSequenceEnvelopeException(0); } - @Nonnull - private static ABIVersion selectVersion(final ConnectClientRequest message) { + private static @NonNull ABIVersion selectVersion(final ConnectClientRequest message) { final Range clientRange = Range.closed(message.getMinVersion(), message.getMaxVersion()); for (ABIVersion v : SUPPORTED_ABIVERSIONS) { if (clientRange.contains(v)) { @@ -543,8 +543,7 @@ public class Shard extends RaftActor { } } - @Nullable - private RequestSuccess handleRequest(final RequestEnvelope envelope, final long now) + private @Nullable RequestSuccess handleRequest(final RequestEnvelope envelope, final long now) throws RequestException { // We are not the leader, hence we want to fail-fast. if (!isLeader() || paused || !isLeaderActive()) { @@ -586,6 +585,10 @@ public class Shard extends RaftActor { return roleChangeNotifier; } + String getShardName() { + return shardName; + } + @Override protected LeaderStateChanged newLeaderStateChanged(final String memberId, final String leaderId, final short leaderPayloadVersion) { @@ -754,7 +757,8 @@ public class Shard extends RaftActor { LOG.debug("{}: Forwarding ForwardedReadyTransaction to leader {}", persistenceId(), leader); ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(forwardedReady.getTransactionId(), - forwardedReady.getTransaction().getSnapshot(), forwardedReady.isDoImmediateCommit()); + forwardedReady.getTransaction().getSnapshot(), forwardedReady.isDoImmediateCommit(), + forwardedReady.getParticipatingShardNames()); readyLocal.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion()); leader.forward(readyLocal, getContext()); } @@ -781,9 +785,16 @@ public class Shard extends RaftActor { } private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) { - final LocalHistoryIdentifier id = closeTransactionChain.getIdentifier(); - store.closeTransactionChain(id, null); - store.purgeTransactionChain(id, null); + if (isLeader()) { + final LocalHistoryIdentifier id = closeTransactionChain.getIdentifier(); + // FIXME: CONTROLLER-1628: stage purge once no transactions are present + store.closeTransactionChain(id, null); + store.purgeTransactionChain(id, null); + } else if (getLeader() != null) { + getLeader().forward(closeTransactionChain, getContext()); + } else { + LOG.warn("{}: Could not close transaction {}", persistenceId(), closeTransactionChain.getIdentifier()); + } } @SuppressWarnings("checkstyle:IllegalCatch") @@ -831,7 +842,6 @@ public class Shard extends RaftActor { } @Override - @Nonnull protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() { if (restoreFromSnapshot == null) { return ShardRecoveryCoordinator.create(store, persistenceId(), LOG); @@ -851,7 +861,7 @@ public class Shard extends RaftActor { if (txCommitTimeoutCheckSchedule == null) { // Schedule a message to be periodically sent to check if the current in-progress // transaction should be expired and aborted. - FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS); + FiniteDuration period = FiniteDuration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS); txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule( period, period, getSelf(), TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender()); @@ -923,6 +933,8 @@ public class Shard extends RaftActor { messagesToForward.size(), leader); for (Object message : messagesToForward) { + LOG.debug("{}: Forwarding pending transaction message {}", persistenceId(), message); + leader.tell(message, self()); } }