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%2Fentityownership%2FEntityOwnershipShardCommitCoordinator.java;h=f3adfbd51e373a0f9c8a515040123f71e735a448;hb=634dfac8eead60f443bf75e749c70d1f2bb29198;hp=c792cf1dda996370b52002fa952b5689088d73df;hpb=07c96b0fa318b7bf559df4954f705d06a44f1354;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardCommitCoordinator.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardCommitCoordinator.java index c792cf1dda..f3adfbd51e 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardCommitCoordinator.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardCommitCoordinator.java @@ -13,8 +13,10 @@ 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; @@ -62,13 +64,13 @@ class EntityOwnershipShardCommitCoordinator { 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; @@ -79,12 +81,12 @@ class EntityOwnershipShardCommitCoordinator { 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 { @@ -94,13 +96,13 @@ class EntityOwnershipShardCommitCoordinator { 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(); @@ -113,7 +115,7 @@ class EntityOwnershipShardCommitCoordinator { 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()); @@ -121,63 +123,62 @@ class EntityOwnershipShardCommitCoordinator { 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 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 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"); } - pendingModifications.addAll(modifications.getModifications()); + pendingModifications.addAll(modifications); } else { - inflightCommit = modifications; + inflightCommit = newBatchedModifications(); + inflightCommit.addModifications(modifications); shard.tryCommitModifications(inflightCommit); } } @@ -187,12 +188,12 @@ class EntityOwnershipShardCommitCoordinator { possiblyPrunePendingCommits(shard, isLeader); - if(!isLeader && inflightCommit != null) { + 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(); } @@ -216,29 +217,26 @@ class EntityOwnershipShardCommitCoordinator { shard.convertPendingTransactionsToMessages(); // Prune the inflightCommit. - if(inflightCommit != null) { + if (inflightCommit != null) { inflightCommit = pruneModifications(inflightCommit); } // Prune the subsequent pending modifications. - Iterator 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) { - BatchedModifications prunedModifications = new BatchedModifications(toPrune.getTransactionID(), toPrune.getVersion()); + 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)) { + for (Modification mod: toPrune.getModifications()) { + if (canForwardModificationToNewLeader(mod)) { prunedModifications.addModification(mod); } } @@ -265,15 +263,15 @@ class EntityOwnershipShardCommitCoordinator { 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( new TransactionIdentifier(historyId, ++transactionIDCounter), DataStoreVersions.CURRENT_VERSION); modifications.setDoCommitOnReady(true); - modifications.setReady(true); + modifications.setReady(); modifications.setTotalMessagesSent(1); return modifications; }