Renew all the leases at once, rather than one at a time

Author: DaveCTurner

server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java

@@ -60,6 +60,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 /**
  * This class is responsible for tracking the replication group with its progress and safety markers (local and global checkpoints).
@@ -462,6 +463,7 @@ public static String getPeerRecoveryRetentionLeaseId(ShardRouting shardRouting)
      */
     public synchronized void renewPeerRecoveryRetentionLeases() {
         assert primaryMode;
+        assert invariant();
 
         /*
          * Peer-recovery retention leases never expire while the associated shard is assigned, but we must still renew them occasionally in
@@ -471,28 +473,40 @@ public synchronized void renewPeerRecoveryRetentionLeases() {
          */
         final long renewalTimeMillis = currentTimeMillisSupplier.getAsLong() - indexSettings.getRetentionLeaseMillis() / 2;
 
-        for (ShardRouting shardRouting : routingTable) {
-            if (shardRouting.assignedToNode()) {
-                final CheckpointState checkpointState = checkpoints.get(shardRouting.allocationId().getId());
+        /*
+         * If any of the peer-recovery retention leases need renewal, it's a good opportunity to renew them all.
+         */
+        final boolean renewalNeeded = StreamSupport.stream(routingTable.spliterator(), false).filter(ShardRouting::assignedToNode)
+            .anyMatch(shardRouting -> {
                 final RetentionLease retentionLease = retentionLeases.get(getPeerRecoveryRetentionLeaseId(shardRouting));
                 if (retentionLease == null) {
-                    if (checkpointState.tracked) {
-                        /*
-                         * BWC: We got here here via a rolling upgrade from an older version that doesn't create peer recovery retention
-                         * leases for every shard copy. TODO create leases lazily
-                         */
-                        assert indexSettings.getIndexVersionCreated().before(Version.V_8_0_0) : indexSettings.getIndexVersionCreated();
-                    }
-                } else {
-                    if (retentionLease.retainingSequenceNumber() <= checkpointState.globalCheckpoint
-                          || retentionLease.timestamp() <= renewalTimeMillis) {
+                    /*
+                     * If this shard copy is tracked then we got here here via a rolling upgrade from an older version that doesn't
+                     * create peer recovery retention leases for every shard copy. TODO create leases lazily in that situation.
+                     */
+                    assert checkpoints.get(shardRouting.allocationId().getId()).tracked == false
+                        || indexSettings.getIndexVersionCreated().before(Version.V_8_0_0);
+                    return false;
+                }
+                return retentionLease.timestamp() <= renewalTimeMillis
+                    || retentionLease.retainingSequenceNumber() <= checkpoints.get(shardRouting.allocationId().getId()).globalCheckpoint;
+            });
+
+        if (renewalNeeded) {
+            for (ShardRouting shardRouting : routingTable) {
+                if (shardRouting.assignedToNode()) {
+                    final RetentionLease retentionLease = retentionLeases.get(getPeerRecoveryRetentionLeaseId(shardRouting));
+                    if (retentionLease != null) {
+                        final CheckpointState checkpointState = checkpoints.get(shardRouting.allocationId().getId());
                         renewRetentionLease(getPeerRecoveryRetentionLeaseId(shardRouting),
                             Math.max(0L, checkpointState.globalCheckpoint + 1L),
                             PEER_RECOVERY_RETENTION_LEASE_SOURCE);
                     }
                 }
             }
         }
+
+        assert invariant();
     }
 
     public static class CheckpointState implements Writeable {

server/src/test/java/org/elasticsearch/index/seqno/ReplicationTrackerTests.java

@@ -1094,12 +1094,6 @@ public void testPeerRecoveryRetentionLeaseCreationAndRenewal() {
         tracker.renewPeerRecoveryRetentionLeases();
         assertTrue("expired extra lease", tracker.getRetentionLeases(true).v1());
 
-
-        for (RetentionLease retentionLease : tracker.getRetentionLeases().leases()) {
-            // update all leases' timestamps so they don't need a time-based renewal for a while
-            tracker.renewRetentionLease(retentionLease.id(), retentionLease.retainingSequenceNumber(), retentionLease.source());
-        }
-
         final AllocationId advancingAllocationId
             = initializingAllocationIds.isEmpty() || rarely() ? primaryId : randomFrom(initializingAllocationIds);
         final String advancingLeaseId = retentionLeaseFromAllocationId.apply(advancingAllocationId).id();
@@ -1117,14 +1111,24 @@ public void testPeerRecoveryRetentionLeaseCreationAndRenewal() {
         tracker.renewPeerRecoveryRetentionLeases();
         assertThat("immediate renewal is a no-op", tracker.getRetentionLeases().version(), equalTo(initialVersion));
 
-        final long shorterThanRenewalTime = randomLongBetween(0L, peerRecoveryRetentionLeaseRenewalTimeMillis - 1);
-        currentTimeMillis.addAndGet(shorterThanRenewalTime);
-        tracker.renewPeerRecoveryRetentionLeases();
-        assertThat("renewal is a no-op after a short time", tracker.getRetentionLeases().version(), equalTo(initialVersion));
+        //noinspection OptionalGetWithoutIsPresent
+        final long millisUntilFirstRenewal
+            = tracker.getRetentionLeases().leases().stream().mapToLong(RetentionLease::timestamp).min().getAsLong()
+            + peerRecoveryRetentionLeaseRenewalTimeMillis
+            - currentTimeMillis.get();
+
+        if (millisUntilFirstRenewal != 0) {
+            final long shorterThanRenewalTime = randomLongBetween(0L, millisUntilFirstRenewal - 1);
+            currentTimeMillis.addAndGet(shorterThanRenewalTime);
+            tracker.renewPeerRecoveryRetentionLeases();
+            assertThat("renewal is a no-op after a short time", tracker.getRetentionLeases().version(), equalTo(initialVersion));
+            currentTimeMillis.addAndGet(millisUntilFirstRenewal - shorterThanRenewalTime);
+        }
 
-        currentTimeMillis.addAndGet(peerRecoveryRetentionLeaseRenewalTimeMillis - shorterThanRenewalTime);
         tracker.renewPeerRecoveryRetentionLeases();
         assertThat("renewal happens after a sufficiently long time", tracker.getRetentionLeases().version(), greaterThan(initialVersion));
+        assertTrue("all leases were renewed",
+            tracker.getRetentionLeases().leases().stream().allMatch(l -> l.timestamp() == currentTimeMillis.get()));
 
         assertThat("test ran for too long, potentially leading to overflow",
             currentTimeMillis.get(), lessThanOrEqualTo(testStartTimeMillis + maximumTestTimeMillis));