+ 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.
+ Iterator<Modification> iter = pendingModifications.iterator();
+ while(iter.hasNext()) {
+ Modification mod = iter.next();
+ if(!canForwardModificationToNewLeader(mod)) {
+ iter.remove();
+ }
+ }
+ }
+ }
+
+ @Nullable
+ private BatchedModifications pruneModifications(BatchedModifications toPrune) {
+ BatchedModifications prunedModifications = new BatchedModifications(toPrune.getTransactionID(), toPrune.getVersion());
+ prunedModifications.setDoCommitOnReady(toPrune.isDoCommitOnReady());
+ prunedModifications.setReady(toPrune.isReady());
+ 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;
+ }
+