*/
package org.opendaylight.controller.cluster.datastore.entityownership;
+import static java.util.Objects.requireNonNull;
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.eclipse.jdt.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;
private Cancellable retryCommitSchedule;
private long transactionIDCounter = 0;
- EntityOwnershipShardCommitCoordinator(MemberName localMemberName, Logger log) {
- this.log = Preconditions.checkNotNull(log);
+ EntityOwnershipShardCommitCoordinator(final MemberName localMemberName, final Logger log) {
+ this.log = requireNonNull(log);
historyId = new LocalHistoryIdentifier(
ClientIdentifier.create(FrontendIdentifier.create(localMemberName, FRONTEND_TYPE), 0), 0);
}
- boolean handleMessage(Object message, EntityOwnershipShard shard) {
+ boolean handleMessage(final Object message, final EntityOwnershipShard shard) {
boolean handled = true;
if (CommitTransactionReply.isSerializedType(message)) {
// Successful reply from a local commit.
return handled;
}
- private void retryInflightCommit(EntityOwnershipShard shard) {
+ private void retryInflightCommit(final EntityOwnershipShard shard) {
// Shouldn't be null happen but verify anyway
if (inflightCommit == null) {
return;
}
}
- void inflightCommitFailure(Throwable cause, EntityOwnershipShard shard) {
+ void inflightCommitFailure(final Throwable cause, final EntityOwnershipShard shard) {
// This should've originated from a failed inflight commit but verify anyway
if (inflightCommit == null) {
return;
scheduleInflightCommitRetry(shard);
}
- private void scheduleInflightCommitRetry(EntityOwnershipShard shard) {
+ private void scheduleInflightCommitRetry(final EntityOwnershipShard shard) {
FiniteDuration duration = shard.getDatastoreContext().getShardRaftConfig().getElectionTimeOutInterval();
log.debug("Scheduling retry for BatchedModifications commit {} in {}",
COMMIT_RETRY_MESSAGE, shard.getContext().dispatcher(), ActorRef.noSender());
}
- void inflightCommitSucceeded(EntityOwnershipShard shard) {
+ void inflightCommitSucceeded(final EntityOwnershipShard shard) {
// Shouldn't be null but verify anyway
if (inflightCommit == null) {
return;
commitNextBatch(shard);
}
- void commitNextBatch(EntityOwnershipShard shard) {
+ void commitNextBatch(final EntityOwnershipShard shard) {
if (inflightCommit != null || pendingModifications.isEmpty() || !shard.hasLeader()) {
return;
}
shard.tryCommitModifications(inflightCommit);
}
- void commitModification(Modification modification, EntityOwnershipShard shard) {
- BatchedModifications modifications = newBatchedModifications();
- modifications.addModification(modification);
- commitModifications(modifications, shard);
+ void commitModification(final Modification modification, final EntityOwnershipShard shard) {
+ commitModifications(ImmutableList.of(modification), shard);
}
- void commitModifications(BatchedModifications modifications, EntityOwnershipShard shard) {
- if (modifications.getModifications().isEmpty()) {
+ void commitModifications(final List<Modification> modifications, final EntityOwnershipShard shard) {
+ if (modifications.isEmpty()) {
return;
}
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) {
+ void onStateChanged(final EntityOwnershipShard shard, final boolean isLeader) {
shard.possiblyRemoveAllInitialCandidates(shard.getLeader());
possiblyPrunePendingCommits(shard, isLeader);
}
}
- private void possiblyPrunePendingCommits(EntityOwnershipShard shard, boolean isLeader) {
+ private void possiblyPrunePendingCommits(final EntityOwnershipShard shard, final 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
}
// Prune the subsequent pending modifications.
- Iterator<Modification> iter = pendingModifications.iterator();
- while (iter.hasNext()) {
- Modification mod = iter.next();
- if (!canForwardModificationToNewLeader(mod)) {
- iter.remove();
- }
- }
+ pendingModifications.removeIf(mod -> !canForwardModificationToNewLeader(mod));
}
}
- @Nullable
- private BatchedModifications pruneModifications(BatchedModifications toPrune) {
+ private @Nullable BatchedModifications pruneModifications(final BatchedModifications toPrune) {
BatchedModifications prunedModifications = new BatchedModifications(toPrune.getTransactionId(),
toPrune.getVersion());
prunedModifications.setDoCommitOnReady(toPrune.isDoCommitOnReady());
- prunedModifications.setReady(toPrune.isReady());
+ if (toPrune.isReady()) {
+ prunedModifications.setReady(toPrune.getParticipatingShardNames());
+ }
prunedModifications.setTotalMessagesSent(toPrune.getTotalMessagesSent());
for (Modification mod: toPrune.getModifications()) {
if (canForwardModificationToNewLeader(mod)) {
return !prunedModifications.getModifications().isEmpty() ? prunedModifications : null;
}
- private boolean canForwardModificationToNewLeader(Modification mod) {
+ private boolean canForwardModificationToNewLeader(final 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) {
inflightCommit = newBatchedModifications;
}
- BatchedModifications newBatchedModifications() {
+ private BatchedModifications newBatchedModifications() {
BatchedModifications modifications = new BatchedModifications(
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