Introduce retention lease expiration (#37195)

This commit implements a straightforward approach to retention lease
expiration. Namely, we inspect which leases are expired when obtaining
the current leases through the replication tracker. At that moment, we
clean the map that persists the retention leases in memory.

Author: jasontedor

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -133,6 +133,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         IndexSettings.INDEX_GC_DELETES_SETTING,
         IndexSettings.INDEX_SOFT_DELETES_SETTING,
         IndexSettings.INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING,
+        IndexSettings.INDEX_SOFT_DELETES_RETENTION_LEASE_SETTING,
         IndicesRequestCache.INDEX_CACHE_REQUEST_ENABLED_SETTING,
         UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING,
         EnableAllocationDecider.INDEX_ROUTING_REBALANCE_ENABLE_SETTING,

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -256,6 +256,17 @@ public final class IndexSettings {
         Setting.longSetting("index.soft_deletes.retention.operations", 0, 0,
             Property.IndexScope, Property.Dynamic);
 
+    /**
+     * Controls the maximum length of time since a retention lease is created or renewed before it is considered expired.
+     */
+    public static final Setting<TimeValue> INDEX_SOFT_DELETES_RETENTION_LEASE_SETTING =
+            Setting.timeSetting(
+                    "index.soft_deletes.retention.lease",
+                    TimeValue.timeValueHours(12),
+                    TimeValue.ZERO,
+                    Property.Dynamic,
+                    Property.IndexScope);
+
     /**
      * The maximum number of refresh listeners allows on this shard.
      */
@@ -316,6 +327,18 @@ public final class IndexSettings {
     private long gcDeletesInMillis = DEFAULT_GC_DELETES.millis();
     private final boolean softDeleteEnabled;
     private volatile long softDeleteRetentionOperations;
+
+    private volatile long retentionLeaseMillis;
+
+    /**
+     * The maximum age of a retention lease before it is considered expired.
+     *
+     * @return the maximum age
+     */
+    public long getRetentionLeaseMillis() {
+        return retentionLeaseMillis;
+    }
+
     private volatile boolean warmerEnabled;
     private volatile int maxResultWindow;
     private volatile int maxInnerResultWindow;
@@ -431,6 +454,7 @@ public IndexSettings(final IndexMetaData indexMetaData, final Settings nodeSetti
         gcDeletesInMillis = scopedSettings.get(INDEX_GC_DELETES_SETTING).getMillis();
         softDeleteEnabled = version.onOrAfter(Version.V_6_5_0) && scopedSettings.get(INDEX_SOFT_DELETES_SETTING);
         softDeleteRetentionOperations = scopedSettings.get(INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING);
+        retentionLeaseMillis = scopedSettings.get(INDEX_SOFT_DELETES_RETENTION_LEASE_SETTING).millis();
         warmerEnabled = scopedSettings.get(INDEX_WARMER_ENABLED_SETTING);
         maxResultWindow = scopedSettings.get(MAX_RESULT_WINDOW_SETTING);
         maxInnerResultWindow = scopedSettings.get(MAX_INNER_RESULT_WINDOW_SETTING);

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

@@ -35,7 +35,6 @@
 import org.elasticsearch.index.shard.ShardId;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -137,6 +136,12 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
      */
     private final LongConsumer onGlobalCheckpointUpdated;
 
+    /**
+     * A supplier of the current time. This supplier is used to add a timestamp to retention leases, and to determine retention lease
+     * expiration.
+     */
+    private final LongSupplier currentTimeMillisSupplier;
+
     /**
      * This set contains allocation IDs for which there is a thread actively waiting for the local checkpoint to advance to at least the
      * current global checkpoint.
@@ -151,12 +156,21 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
     private final Map<String, RetentionLease> retentionLeases = new HashMap<>();
 
     /**
-     * Get all retention leases tracker on this shard. An unmodifiable copy of the retention leases is returned.
+     * Get all non-expired retention leases tracker on this shard. An unmodifiable copy of the retention leases is returned.
      *
      * @return the retention leases
      */
     public synchronized Collection<RetentionLease> getRetentionLeases() {
-        return Collections.unmodifiableCollection(new ArrayList<>(retentionLeases.values()));
+        final long currentTimeMillis = currentTimeMillisSupplier.getAsLong();
+        final long retentionLeaseMillis = indexSettings.getRetentionLeaseMillis();
+        final Collection<RetentionLease> nonExpiredRetentionLeases = retentionLeases
+                .values()
+                .stream()
+                .filter(retentionLease -> currentTimeMillis - retentionLease.timestamp() <= retentionLeaseMillis)
+                .collect(Collectors.toList());
+        retentionLeases.clear();
+        retentionLeases.putAll(nonExpiredRetentionLeases.stream().collect(Collectors.toMap(RetentionLease::id, lease -> lease)));
+        return Collections.unmodifiableCollection(nonExpiredRetentionLeases);
     }
 
     /**
@@ -168,7 +182,7 @@ public synchronized Collection<RetentionLease> getRetentionLeases() {
      */
     public synchronized void addOrUpdateRetentionLease(final String id, final long retainingSequenceNumber, final String source) {
         assert primaryMode;
-        retentionLeases.put(id, new RetentionLease(id, retainingSequenceNumber, source));
+        retentionLeases.put(id, new RetentionLease(id, retainingSequenceNumber, currentTimeMillisSupplier.getAsLong(), source));
     }
 
     public static class CheckpointState implements Writeable {
@@ -425,7 +439,8 @@ public ReplicationTracker(
             final String allocationId,
             final IndexSettings indexSettings,
             final long globalCheckpoint,
-            final LongConsumer onGlobalCheckpointUpdated) {
+            final LongConsumer onGlobalCheckpointUpdated,
+            final LongSupplier currentTimeMillisSupplier) {
         super(shardId, indexSettings);
         assert globalCheckpoint >= SequenceNumbers.UNASSIGNED_SEQ_NO : "illegal initial global checkpoint: " + globalCheckpoint;
         this.shardAllocationId = allocationId;
@@ -435,6 +450,7 @@ public ReplicationTracker(
         this.checkpoints = new HashMap<>(1 + indexSettings.getNumberOfReplicas());
         checkpoints.put(allocationId, new CheckpointState(SequenceNumbers.UNASSIGNED_SEQ_NO, globalCheckpoint, false, false));
         this.onGlobalCheckpointUpdated = Objects.requireNonNull(onGlobalCheckpointUpdated);
+        this.currentTimeMillisSupplier = Objects.requireNonNull(currentTimeMillisSupplier);
         this.pendingInSync = new HashSet<>();
         this.routingTable = null;
         this.replicationGroup = null;

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

@@ -23,9 +23,9 @@
  * A "shard history retention lease" (or "retention lease" for short) is conceptually a marker containing a retaining sequence number such
  * that all operations with sequence number at least that retaining sequence number will be retained during merge operations (which could
  * otherwise merge away operations that have been soft deleted). Each retention lease contains a unique identifier, the retaining sequence
- * number, and the source of the retention lease (e.g., "ccr").
+ * number, the timestamp of when the lease was created or renewed, and the source of the retention lease (e.g., "ccr").
  */
-public class RetentionLease {
+public final class RetentionLease {
 
     private final String id;
 
@@ -50,6 +50,17 @@ public long retainingSequenceNumber() {
         return retainingSequenceNumber;
     }
 
+    private final long timestamp;
+
+    /**
+     * The timestamp of when this retention lease was created or renewed.
+     *
+     * @return the timestamp used as a basis for determining lease expiration
+     */
+    public long timestamp() {
+        return timestamp;
+    }
+
     private final String source;
 
     /**
@@ -66,19 +77,22 @@ public String source() {
      *
      * @param id                      the identifier of the retention lease
      * @param retainingSequenceNumber the retaining sequence number
+     * @param timestamp               the timestamp of when the retention lease was created or renewed
      * @param source                  the source of the retention lease
      */
-    public RetentionLease(final String id, final long retainingSequenceNumber, final String source) {
+    public RetentionLease(final String id, final long retainingSequenceNumber, final long timestamp, final String source) {
         this.id = id;
         this.retainingSequenceNumber = retainingSequenceNumber;
+        this.timestamp = timestamp;
         this.source = source;
     }
 
     @Override
     public String toString() {
-        return "ShardHistoryRetentionLease{" +
+        return "RetentionLease{" +
                 "id='" + id + '\'' +
                 ", retainingSequenceNumber=" + retainingSequenceNumber +
+                ", timestamp=" + timestamp +
                 ", source='" + source + '\'' +
                 '}';
     }

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -305,7 +305,13 @@ public IndexShard(
         this.globalCheckpointListeners =
                 new GlobalCheckpointListeners(shardId, threadPool.executor(ThreadPool.Names.LISTENER), threadPool.scheduler(), logger);
         this.replicationTracker =
-                new ReplicationTracker(shardId, aId, indexSettings, UNASSIGNED_SEQ_NO, globalCheckpointListeners::globalCheckpointUpdated);
+                new ReplicationTracker(
+                        shardId,
+                        aId,
+                        indexSettings,
+                        UNASSIGNED_SEQ_NO,
+                        globalCheckpointListeners::globalCheckpointUpdated,
+                        threadPool::absoluteTimeInMillis);
 
         // the query cache is a node-level thing, however we want the most popular filters
         // to be computed on a per-shard basis

server/src/test/java/org/elasticsearch/index/engine/SoftDeletesPolicyTests.java

@@ -55,7 +55,7 @@ public void testSoftDeletesRetentionLock() {
                 () -> {
                     final Set<RetentionLease> leases = new HashSet<>(retainingSequenceNumbers.length);
                     for (int i = 0; i < retainingSequenceNumbers.length; i++) {
-                        leases.add(new RetentionLease(Integer.toString(i), retainingSequenceNumbers[i].get(), "test"));
+                        leases.add(new RetentionLease(Integer.toString(i), retainingSequenceNumbers[i].get(), 0L, "test"));
                     }
                     return leases;
                 };

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

@@ -21,18 +21,23 @@
 
 import org.elasticsearch.cluster.routing.AllocationId;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.IndexSettingsModule;
 
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongSupplier;
 
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
 public class ReplicationTrackerRetentionLeaseTests extends ReplicationTrackerTestCase {
 
@@ -43,7 +48,8 @@ public void testAddOrUpdateRetentionLease() {
                 id.getId(),
                 IndexSettingsModule.newIndexSettings("test", Settings.EMPTY),
                 UNASSIGNED_SEQ_NO,
-                value -> {});
+                value -> {},
+                () -> 0L);
         replicationTracker.updateFromMaster(
                 randomNonNegativeLong(),
                 Collections.singleton(id.getId()),
@@ -55,19 +61,73 @@ public void testAddOrUpdateRetentionLease() {
         for (int i = 0; i < length; i++) {
             minimumRetainingSequenceNumbers[i] = randomLongBetween(SequenceNumbers.NO_OPS_PERFORMED, Long.MAX_VALUE);
             replicationTracker.addOrUpdateRetentionLease(Integer.toString(i), minimumRetainingSequenceNumbers[i], "test-" + i);
-            assertRetentionLeases(replicationTracker, i + 1, minimumRetainingSequenceNumbers);
+            assertRetentionLeases(replicationTracker, i + 1, minimumRetainingSequenceNumbers, () -> 0L);
         }
 
         for (int i = 0; i < length; i++) {
             minimumRetainingSequenceNumbers[i] = randomLongBetween(minimumRetainingSequenceNumbers[i], Long.MAX_VALUE);
             replicationTracker.addOrUpdateRetentionLease(Integer.toString(i), minimumRetainingSequenceNumbers[i], "test-" + i);
-            assertRetentionLeases(replicationTracker, length, minimumRetainingSequenceNumbers);
+            assertRetentionLeases(replicationTracker, length, minimumRetainingSequenceNumbers, () -> 0L);
         }
 
     }
 
+    public void testExpiration() {
+        final AllocationId id = AllocationId.newInitializing();
+        final AtomicLong currentTimeMillis = new AtomicLong(randomLongBetween(0, 1024));
+        final long retentionLeaseMillis = randomLongBetween(1, TimeValue.timeValueHours(12).millis());
+        final Settings settings = Settings
+                .builder()
+                .put(IndexSettings.INDEX_SOFT_DELETES_RETENTION_LEASE_SETTING.getKey(), TimeValue.timeValueMillis(retentionLeaseMillis))
+                .build();
+        final ReplicationTracker replicationTracker = new ReplicationTracker(
+                new ShardId("test", "_na", 0),
+                id.getId(),
+                IndexSettingsModule.newIndexSettings("test", settings),
+                UNASSIGNED_SEQ_NO,
+                value -> {},
+                currentTimeMillis::get);
+        replicationTracker.updateFromMaster(
+                randomNonNegativeLong(),
+                Collections.singleton(id.getId()),
+                routingTable(Collections.emptySet(), id),
+                Collections.emptySet());
+        replicationTracker.activatePrimaryMode(SequenceNumbers.NO_OPS_PERFORMED);
+        final long[] retainingSequenceNumbers = new long[1];
+        retainingSequenceNumbers[0] = randomLongBetween(SequenceNumbers.NO_OPS_PERFORMED, Long.MAX_VALUE);
+        replicationTracker.addOrUpdateRetentionLease("0", retainingSequenceNumbers[0], "test-0");
+
+        {
+            final Collection<RetentionLease> retentionLeases = replicationTracker.getRetentionLeases();
+            assertThat(retentionLeases, hasSize(1));
+            final RetentionLease retentionLease = retentionLeases.iterator().next();
+            assertThat(retentionLease.timestamp(), equalTo(currentTimeMillis.get()));
+            assertRetentionLeases(replicationTracker, 1, retainingSequenceNumbers, currentTimeMillis::get);
+        }
+
+        // renew the lease
+        currentTimeMillis.set(currentTimeMillis.get() + randomLongBetween(0, 1024));
+        retainingSequenceNumbers[0] = randomLongBetween(retainingSequenceNumbers[0], Long.MAX_VALUE);
+        replicationTracker.addOrUpdateRetentionLease("0", retainingSequenceNumbers[0], "test-0");
+
+        {
+            final Collection<RetentionLease> retentionLeases = replicationTracker.getRetentionLeases();
+            assertThat(retentionLeases, hasSize(1));
+            final RetentionLease retentionLease = retentionLeases.iterator().next();
+            assertThat(retentionLease.timestamp(), equalTo(currentTimeMillis.get()));
+            assertRetentionLeases(replicationTracker, 1, retainingSequenceNumbers, currentTimeMillis::get);
+        }
+
+        // now force the lease to expire
+        currentTimeMillis.set(currentTimeMillis.get() + randomLongBetween(retentionLeaseMillis, Long.MAX_VALUE - currentTimeMillis.get()));
+        assertRetentionLeases(replicationTracker, 0, retainingSequenceNumbers, currentTimeMillis::get);
+    }
+
     private void assertRetentionLeases(
-            final ReplicationTracker replicationTracker, final int size, final long[] minimumRetainingSequenceNumbers) {
+            final ReplicationTracker replicationTracker,
+            final int size,
+            final long[] minimumRetainingSequenceNumbers,
+            final LongSupplier currentTimeMillisSupplier) {
         final Collection<RetentionLease> retentionLeases = replicationTracker.getRetentionLeases();
         final Map<String, RetentionLease> idToRetentionLease = new HashMap<>();
         for (final RetentionLease retentionLease : retentionLeases) {
@@ -79,6 +139,9 @@ private void assertRetentionLeases(
             assertThat(idToRetentionLease.keySet(), hasItem(Integer.toString(i)));
             final RetentionLease retentionLease = idToRetentionLease.get(Integer.toString(i));
             assertThat(retentionLease.retainingSequenceNumber(), equalTo(minimumRetainingSequenceNumbers[i]));
+            assertThat(
+                    currentTimeMillisSupplier.getAsLong() - retentionLease.timestamp(),
+                    lessThanOrEqualTo(replicationTracker.indexSettings().getRetentionLeaseMillis()));
             assertThat(retentionLease.source(), equalTo("test-" + i));
         }
     }

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

@@ -31,18 +31,23 @@
 
 import java.util.Set;
 import java.util.function.LongConsumer;
+import java.util.function.LongSupplier;
 
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
 
 public abstract class ReplicationTrackerTestCase extends ESTestCase  {
 
-    ReplicationTracker newTracker(final AllocationId allocationId, final LongConsumer updatedGlobalCheckpoint) {
+    ReplicationTracker newTracker(
+            final AllocationId allocationId,
+            final LongConsumer updatedGlobalCheckpoint,
+            final LongSupplier currentTimeMillisSupplier) {
         return new ReplicationTracker(
                 new ShardId("test", "_na_", 0),
                 allocationId.getId(),
                 IndexSettingsModule.newIndexSettings("test", Settings.EMPTY),
                 UNASSIGNED_SEQ_NO,
-                updatedGlobalCheckpoint);
+                updatedGlobalCheckpoint,
+                currentTimeMillisSupplier);
     }
 
     static IndexShardRoutingTable routingTable(final Set<AllocationId> initializingIds, final AllocationId primaryId) {

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

@@ -406,7 +406,7 @@ public void testWaitForAllocationIdToBeInSync() throws Exception {
     private AtomicLong updatedGlobalCheckpoint = new AtomicLong(UNASSIGNED_SEQ_NO);
 
     private ReplicationTracker newTracker(final AllocationId allocationId) {
-        return newTracker(allocationId, updatedGlobalCheckpoint::set);
+        return newTracker(allocationId, updatedGlobalCheckpoint::set, () -> 0L);
     }
 
     public void testWaitForAllocationIdToBeInSyncCanBeInterrupted() throws BrokenBarrierException, InterruptedException {
@@ -683,10 +683,10 @@ public void testPrimaryContextHandoff() throws IOException {
         final AllocationId primaryAllocationId = clusterState.routingTable.primaryShard().allocationId();
         final LongConsumer onUpdate = updatedGlobalCheckpoint -> {};
         final long globalCheckpoint = UNASSIGNED_SEQ_NO;
-        ReplicationTracker oldPrimary =
-                new ReplicationTracker(shardId, primaryAllocationId.getId(), indexSettings, globalCheckpoint, onUpdate);
-        ReplicationTracker newPrimary =
-                new ReplicationTracker(shardId, primaryAllocationId.getRelocationId(), indexSettings, globalCheckpoint, onUpdate);
+        ReplicationTracker oldPrimary = new ReplicationTracker(
+                        shardId, primaryAllocationId.getId(), indexSettings, globalCheckpoint, onUpdate, () -> 0L);
+        ReplicationTracker newPrimary = new ReplicationTracker(
+                shardId, primaryAllocationId.getRelocationId(), indexSettings, globalCheckpoint, onUpdate, () -> 0L);
 
         Set<String> allocationIds = new HashSet<>(Arrays.asList(oldPrimary.shardAllocationId, newPrimary.shardAllocationId));