*/
package org.opendaylight.controller.cluster.datastore.entityownership;
+import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNER_QNAME;
+
import akka.actor.ActorRef;
import akka.actor.Cancellable;
import akka.actor.Status.Failure;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
import java.util.Iterator;
import java.util.LinkedList;
+import java.util.List;
import java.util.Queue;
+import javax.annotation.Nullable;
+import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
+import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
+import org.opendaylight.controller.cluster.access.concepts.FrontendType;
+import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
+import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
-import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
+import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.slf4j.Logger;
import scala.concurrent.duration.FiniteDuration;
return "entityCommitRetry";
}
};
+ private static final FrontendType FRONTEND_TYPE = FrontendType.forName("entity-ownership-internal");
- private final Logger log;
- private int transactionIDCounter = 0;
- private final MemberName localMemberName;
private final Queue<Modification> pendingModifications = new LinkedList<>();
+ private final LocalHistoryIdentifier historyId;
+ private final Logger log;
+
private BatchedModifications inflightCommit;
private Cancellable retryCommitSchedule;
+ private long transactionIDCounter = 0;
EntityOwnershipShardCommitCoordinator(MemberName localMemberName, Logger log) {
- this.localMemberName = localMemberName;
- this.log = log;
+ this.log = Preconditions.checkNotNull(log);
+ historyId = new LocalHistoryIdentifier(
+ ClientIdentifier.create(FrontendIdentifier.create(localMemberName, FRONTEND_TYPE), 0), 0);
}
boolean handleMessage(Object message, EntityOwnershipShard shard) {
boolean handled = true;
- if(CommitTransactionReply.isSerializedType(message)) {
+ if (CommitTransactionReply.isSerializedType(message)) {
// Successful reply from a local commit.
inflightCommitSucceeded(shard);
- } else if(message instanceof akka.actor.Status.Failure) {
+ } else if (message instanceof akka.actor.Status.Failure) {
// Failure reply from a local commit.
- inflightCommitFailure(((Failure)message).cause(), shard);
- } else if(COMMIT_RETRY_MESSAGE.equals(message)) {
+ inflightCommitFailure(((Failure) message).cause(), shard);
+ } else if (COMMIT_RETRY_MESSAGE.equals(message)) {
retryInflightCommit(shard);
} else {
handled = false;
private void retryInflightCommit(EntityOwnershipShard shard) {
// Shouldn't be null happen but verify anyway
- if(inflightCommit == null) {
+ if (inflightCommit == null) {
return;
}
- if(shard.hasLeader()) {
- log.debug("Retrying commit for BatchedModifications {}", inflightCommit.getTransactionID());
+ if (shard.hasLeader()) {
+ log.debug("Retrying commit for BatchedModifications {}", inflightCommit.getTransactionId());
shard.tryCommitModifications(inflightCommit);
} else {
void inflightCommitFailure(Throwable cause, EntityOwnershipShard shard) {
// This should've originated from a failed inflight commit but verify anyway
- if(inflightCommit == null) {
+ if (inflightCommit == null) {
return;
}
- log.debug("Inflight BatchedModifications {} commit failed", inflightCommit.getTransactionID(), cause);
+ log.debug("Inflight BatchedModifications {} commit failed", inflightCommit.getTransactionId(), cause);
- if(!(cause instanceof NoShardLeaderException)) {
+ if (!(cause instanceof NoShardLeaderException)) {
// If the failure is other than NoShardLeaderException the commit may have been partially
// processed so retry with a new transaction ID to be safe.
newInflightCommitWithDifferentTransactionID();
FiniteDuration duration = shard.getDatastoreContext().getShardRaftConfig().getElectionTimeOutInterval();
log.debug("Scheduling retry for BatchedModifications commit {} in {}",
- inflightCommit.getTransactionID(), duration);
+ inflightCommit.getTransactionId(), duration);
retryCommitSchedule = shard.getContext().system().scheduler().scheduleOnce(duration, shard.getSelf(),
COMMIT_RETRY_MESSAGE, shard.getContext().dispatcher(), ActorRef.noSender());
void inflightCommitSucceeded(EntityOwnershipShard shard) {
// Shouldn't be null but verify anyway
- if(inflightCommit == null) {
+ if (inflightCommit == null) {
return;
}
- if(retryCommitSchedule != null) {
+ if (retryCommitSchedule != null) {
retryCommitSchedule.cancel();
}
- log.debug("BatchedModifications commit {} succeeded", inflightCommit.getTransactionID());
+ log.debug("BatchedModifications commit {} succeeded", inflightCommit.getTransactionId());
inflightCommit = null;
commitNextBatch(shard);
}
void commitNextBatch(EntityOwnershipShard shard) {
- if(inflightCommit != null || pendingModifications.isEmpty() || !shard.hasLeader()) {
+ if (inflightCommit != null || pendingModifications.isEmpty() || !shard.hasLeader()) {
return;
}
inflightCommit = newBatchedModifications();
Iterator<Modification> iter = pendingModifications.iterator();
- while(iter.hasNext()) {
+ while (iter.hasNext()) {
inflightCommit.addModification(iter.next());
iter.remove();
- if(inflightCommit.getModifications().size() >=
- shard.getDatastoreContext().getShardBatchedModificationCount()) {
+ if (inflightCommit.getModifications().size()
+ >= shard.getDatastoreContext().getShardBatchedModificationCount()) {
break;
}
}
- log.debug("Committing next BatchedModifications {}, size {}", inflightCommit.getTransactionID(),
+ log.debug("Committing next BatchedModifications {}, size {}", inflightCommit.getTransactionId(),
inflightCommit.getModifications().size());
shard.tryCommitModifications(inflightCommit);
}
void commitModification(Modification modification, EntityOwnershipShard shard) {
- BatchedModifications modifications = newBatchedModifications();
- modifications.addModification(modification);
- commitModifications(modifications, shard);
+ commitModifications(ImmutableList.of(modification), shard);
}
- void commitModifications(BatchedModifications modifications, EntityOwnershipShard shard) {
- if(modifications.getModifications().isEmpty()) {
+ void commitModifications(List<Modification> modifications, EntityOwnershipShard shard) {
+ if (modifications.isEmpty()) {
return;
}
boolean hasLeader = shard.hasLeader();
- if(inflightCommit != null || !hasLeader) {
- if(log.isDebugEnabled()) {
+ if (inflightCommit != null || !hasLeader) {
+ if (log.isDebugEnabled()) {
log.debug("{} - adding modifications to pending",
- (inflightCommit != null ? "A commit is inflight" : "No shard leader"));
+ inflightCommit != null ? "A commit is inflight" : "No shard leader");
}
- pendingModifications.addAll(modifications.getModifications());
+ pendingModifications.addAll(modifications);
} else {
- inflightCommit = modifications;
+ inflightCommit = newBatchedModifications();
+ inflightCommit.addModifications(modifications);
shard.tryCommitModifications(inflightCommit);
}
}
void onStateChanged(EntityOwnershipShard shard, boolean isLeader) {
- if(!isLeader && inflightCommit != null) {
+ shard.possiblyRemoveAllInitialCandidates(shard.getLeader());
+
+ possiblyPrunePendingCommits(shard, isLeader);
+
+ if (!isLeader && inflightCommit != null) {
// We're no longer the leader but we have an inflight local commit. This likely means we didn't get
// consensus for the commit and switched to follower due to another node with a higher term. We
// can't be sure if the commit was replicated to any node so we retry it here with a new
// transaction ID.
- if(retryCommitSchedule != null) {
+ if (retryCommitSchedule != null) {
retryCommitSchedule.cancel();
}
}
}
+ private void possiblyPrunePendingCommits(EntityOwnershipShard shard, boolean isLeader) {
+ // If we were the leader and transitioned to follower, we'll try to forward pending commits to the new leader.
+ // However certain commits, e.g. entity owner changes, should only be committed by a valid leader as the
+ // criteria used to determine the commit may be stale. Since we're no longer a valid leader, we should not
+ // forward such commits thus we prune the pending modifications. We still should forward local candidate change
+ // commits.
+ if (shard.hasLeader() && !isLeader) {
+ // We may have already submitted a transaction for replication and commit. We don't need the base Shard to
+ // forward it since we also have it stored in the inflightCommit and handle retries. So we just clear
+ // pending transactions and drop them.
+ shard.convertPendingTransactionsToMessages();
+
+ // Prune the inflightCommit.
+ if (inflightCommit != null) {
+ inflightCommit = pruneModifications(inflightCommit);
+ }
+
+ // Prune the subsequent pending modifications.
+ pendingModifications.removeIf(mod -> !canForwardModificationToNewLeader(mod));
+ }
+ }
+
+ @Nullable
+ private BatchedModifications pruneModifications(BatchedModifications toPrune) {
+ BatchedModifications prunedModifications = new BatchedModifications(toPrune.getTransactionId(),
+ toPrune.getVersion());
+ prunedModifications.setDoCommitOnReady(toPrune.isDoCommitOnReady());
+ if (toPrune.isReady()) {
+ prunedModifications.setReady(toPrune.getParticipatingShardNames());
+ }
+ prunedModifications.setTotalMessagesSent(toPrune.getTotalMessagesSent());
+ for (Modification mod: toPrune.getModifications()) {
+ if (canForwardModificationToNewLeader(mod)) {
+ prunedModifications.addModification(mod);
+ }
+ }
+
+ return !prunedModifications.getModifications().isEmpty() ? prunedModifications : null;
+ }
+
+ private boolean canForwardModificationToNewLeader(Modification mod) {
+ // If this is a WRITE of entity owner we don't want to forward it to a new leader since the criteria used
+ // to determine the new owner might be stale.
+ if (mod instanceof WriteModification) {
+ WriteModification writeMod = (WriteModification)mod;
+ boolean canForward = !writeMod.getPath().getLastPathArgument().getNodeType().equals(ENTITY_OWNER_QNAME);
+
+ if (!canForward) {
+ log.debug("Not forwarding WRITE modification for {} to new leader", writeMod.getPath());
+ }
+
+ return canForward;
+ }
+
+ return true;
+ }
+
private void newInflightCommitWithDifferentTransactionID() {
BatchedModifications newBatchedModifications = newBatchedModifications();
- newBatchedModifications.getModifications().addAll(inflightCommit.getModifications());
+ newBatchedModifications.addModifications(inflightCommit.getModifications());
inflightCommit = newBatchedModifications;
}
- BatchedModifications newBatchedModifications() {
+ private BatchedModifications newBatchedModifications() {
BatchedModifications modifications = new BatchedModifications(
- TransactionIdentifier.create(localMemberName, ++transactionIDCounter).toString(),
- DataStoreVersions.CURRENT_VERSION, "");
+ new TransactionIdentifier(historyId, ++transactionIDCounter), DataStoreVersions.CURRENT_VERSION);
modifications.setDoCommitOnReady(true);
- modifications.setReady(true);
+ modifications.setReady();
modifications.setTotalMessagesSent(1);
return modifications;
}
Code Review / controller.git / blobdiff