Lift retention lease expiration to index shard (#38380)

This commit lifts the control of when retention leases are expired to
index shard. In this case, we move expiration to an explicit action
rather than a side-effect of calling
ReplicationTracker#getRetentionLeases. This explicit action is invoked
on a timer. If any retention leases expire, then we hard sync the
retention leases to the replicas. Otherwise, we proceed with a
background sync.

Author: jasontedor

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

@@ -121,7 +121,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private volatile AsyncRefreshTask refreshTask;
     private volatile AsyncTranslogFSync fsyncTask;
     private volatile AsyncGlobalCheckpointTask globalCheckpointTask;
-    private volatile AsyncRetentionLeaseBackgroundSyncTask retentionLeaseBackgroundSyncTask;
+    private volatile AsyncRetentionLeaseSyncTask retentionLeaseSyncTask;
 
     // don't convert to Setting<> and register... we only set this in tests and register via a plugin
     private final String INDEX_TRANSLOG_RETENTION_CHECK_INTERVAL_SETTING = "index.translog.retention.check_interval";
@@ -198,7 +198,7 @@ public IndexService(
         this.refreshTask = new AsyncRefreshTask(this);
         this.trimTranslogTask = new AsyncTrimTranslogTask(this);
         this.globalCheckpointTask = new AsyncGlobalCheckpointTask(this);
-        this.retentionLeaseBackgroundSyncTask = new AsyncRetentionLeaseBackgroundSyncTask(this);
+        this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
         rescheduleFsyncTask(indexSettings.getTranslogDurability());
     }
 
@@ -289,7 +289,7 @@ public synchronized void close(final String reason, boolean delete) throws IOExc
                         fsyncTask,
                         trimTranslogTask,
                         globalCheckpointTask,
-                        retentionLeaseBackgroundSyncTask);
+                        retentionLeaseSyncTask);
             }
         }
     }
@@ -788,8 +788,8 @@ private void maybeSyncGlobalCheckpoints() {
         sync(is -> is.maybeSyncGlobalCheckpoint("background"), "global checkpoint");
     }
 
-    private void backgroundSyncRetentionLeases() {
-        sync(IndexShard::backgroundSyncRetentionLeases, "retention lease");
+    private void syncRetentionLeases() {
+        sync(IndexShard::syncRetentionLeases, "retention lease");
     }
 
     private void sync(final Consumer<IndexShard> sync, final String source) {
@@ -812,11 +812,11 @@ private void sync(final Consumer<IndexShard> sync, final String source) {
                                                 && e instanceof IndexShardClosedException == false) {
                                             logger.warn(
                                                     new ParameterizedMessage(
-                                                            "{} failed to execute background {} sync", shard.shardId(), source), e);
+                                                            "{} failed to execute {} sync", shard.shardId(), source), e);
                                         }
                                     },
                                     ThreadPool.Names.SAME,
-                                    "background " + source + " sync");
+                                    source + " sync");
                         } catch (final AlreadyClosedException | IndexShardClosedException e) {
                             // the shard was closed concurrently, continue
                         }
@@ -957,15 +957,15 @@ public String toString() {
         }
     }
 
-    final class AsyncRetentionLeaseBackgroundSyncTask extends BaseAsyncTask {
+    final class AsyncRetentionLeaseSyncTask extends BaseAsyncTask {
 
-        AsyncRetentionLeaseBackgroundSyncTask(final IndexService indexService) {
+        AsyncRetentionLeaseSyncTask(final IndexService indexService) {
             super(indexService, RETENTION_LEASE_SYNC_INTERVAL_SETTING.get(indexService.getIndexSettings().getSettings()));
         }
 
         @Override
         protected void runInternal() {
-            indexService.backgroundSyncRetentionLeases();
+            indexService.syncRetentionLeases();
         }
 
         @Override
@@ -975,7 +975,7 @@ protected String getThreadPool() {
 
         @Override
         public String toString() {
-            return "retention_lease_background_sync";
+            return "retention_lease_sync";
         }
 
     }

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

@@ -339,6 +339,10 @@ public long getRetentionLeaseMillis() {
         return retentionLeaseMillis;
     }
 
+    private void setRetentionLeaseMillis(final TimeValue retentionLease) {
+        this.retentionLeaseMillis = retentionLease.millis();
+    }
+
     private volatile boolean warmerEnabled;
     private volatile int maxResultWindow;
     private volatile int maxInnerResultWindow;
@@ -523,6 +527,7 @@ public IndexSettings(final IndexMetaData indexMetaData, final Settings nodeSetti
         scopedSettings.addSettingsUpdateConsumer(DEFAULT_PIPELINE, this::setDefaultPipeline);
         scopedSettings.addSettingsUpdateConsumer(INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING, this::setSoftDeleteRetentionOperations);
         scopedSettings.addSettingsUpdateConsumer(INDEX_SEARCH_THROTTLED, this::setSearchThrottled);
+        scopedSettings.addSettingsUpdateConsumer(INDEX_SOFT_DELETES_RETENTION_LEASE_SETTING, this::setRetentionLeaseMillis);
     }
 
     private void setSearchIdleAfter(TimeValue searchIdleAfter) { this.searchIdleAfter = searchIdleAfter; }

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

@@ -28,6 +28,7 @@
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.SuppressForbidden;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -155,10 +156,10 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
     private final LongSupplier currentTimeMillisSupplier;
 
     /**
-     * A callback when a new retention lease is created or an existing retention lease expires. In practice, this callback invokes the
-     * retention lease sync action, to sync retention leases to replicas.
+     * A callback when a new retention lease is created. In practice, this callback invokes the retention lease sync action, to sync
+     * retention leases to replicas.
      */
-    private final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onSyncRetentionLeases;
+    private final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onAddRetentionLease;
 
     /**
      * This set contains allocation IDs for which there is a thread actively waiting for the local checkpoint to advance to at least the
@@ -177,43 +178,42 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
     private RetentionLeases retentionLeases = RetentionLeases.EMPTY;
 
     /**
-     * Get all non-expired retention leases tracked on this shard. Note that only the primary shard calculates which leases are expired,
-     * and if any have expired, syncs the retention leases to any replicas.
+     * Get all retention leases tracked on this shard.
      *
      * @return the retention leases
      */
     public RetentionLeases getRetentionLeases() {
-        final boolean wasPrimaryMode;
-        final RetentionLeases nonExpiredRetentionLeases;
-        synchronized (this) {
-            if (primaryMode) {
-                // the primary calculates the non-expired retention leases and syncs them to replicas
-                final long currentTimeMillis = currentTimeMillisSupplier.getAsLong();
-                final long retentionLeaseMillis = indexSettings.getRetentionLeaseMillis();
-                final Map<Boolean, List<RetentionLease>> partitionByExpiration = retentionLeases
-                        .leases()
-                        .stream()
-                        .collect(Collectors.groupingBy(lease -> currentTimeMillis - lease.timestamp() > retentionLeaseMillis));
-                if (partitionByExpiration.get(true) == null) {
-                    // early out as no retention leases have expired
-                    return retentionLeases;
-                }
-                final Collection<RetentionLease> nonExpiredLeases =
-                        partitionByExpiration.get(false) != null ? partitionByExpiration.get(false) : Collections.emptyList();
-                retentionLeases = new RetentionLeases(operationPrimaryTerm, retentionLeases.version() + 1, nonExpiredLeases);
-            }
-            /*
-             * At this point, we were either in primary mode and have updated the non-expired retention leases into the tracking map, or
-             * we were in replica mode and merely need to copy the existing retention leases since a replica does not calculate the
-             * non-expired retention leases, instead receiving them on syncs from the primary.
-             */
-            wasPrimaryMode = primaryMode;
-            nonExpiredRetentionLeases = retentionLeases;
+        return getRetentionLeases(false).v2();
+    }
+
+    /**
+     * If the expire leases parameter is false, gets all retention leases tracked on this shard and otherwise first calculates
+     * expiration of existing retention leases, and then gets all non-expired retention leases tracked on this shard. Note that only the
+     * primary shard calculates which leases are expired, and if any have expired, syncs the retention leases to any replicas. If the
+     * expire leases parameter is true, this replication tracker must be in primary mode.
+     *
+     * @return a tuple indicating whether or not any retention leases were expired, and the non-expired retention leases
+     */
+    public synchronized Tuple<Boolean, RetentionLeases> getRetentionLeases(final boolean expireLeases) {
+        if (expireLeases == false) {
+            return Tuple.tuple(false, retentionLeases);
         }
-        if (wasPrimaryMode) {
-            onSyncRetentionLeases.accept(nonExpiredRetentionLeases, ActionListener.wrap(() -> {}));
+        assert primaryMode;
+        // the primary calculates the non-expired retention leases and syncs them to replicas
+        final long currentTimeMillis = currentTimeMillisSupplier.getAsLong();
+        final long retentionLeaseMillis = indexSettings.getRetentionLeaseMillis();
+        final Map<Boolean, List<RetentionLease>> partitionByExpiration = retentionLeases
+                .leases()
+                .stream()
+                .collect(Collectors.groupingBy(lease -> currentTimeMillis - lease.timestamp() > retentionLeaseMillis));
+        if (partitionByExpiration.get(true) == null) {
+            // early out as no retention leases have expired
+            return Tuple.tuple(false, retentionLeases);
         }
-        return nonExpiredRetentionLeases;
+        final Collection<RetentionLease> nonExpiredLeases =
+                partitionByExpiration.get(false) != null ? partitionByExpiration.get(false) : Collections.emptyList();
+        retentionLeases = new RetentionLeases(operationPrimaryTerm, retentionLeases.version() + 1, nonExpiredLeases);
+        return Tuple.tuple(true, retentionLeases);
     }
 
     /**
@@ -246,7 +246,7 @@ public RetentionLease addRetentionLease(
                     Stream.concat(retentionLeases.leases().stream(), Stream.of(retentionLease)).collect(Collectors.toList()));
             currentRetentionLeases = retentionLeases;
         }
-        onSyncRetentionLeases.accept(currentRetentionLeases, listener);
+        onAddRetentionLease.accept(currentRetentionLeases, listener);
         return retentionLease;
     }
 
@@ -563,7 +563,7 @@ private static long inSyncCheckpointStates(
      * @param indexSettings         the index settings
      * @param operationPrimaryTerm  the current primary term
      * @param globalCheckpoint      the last known global checkpoint for this shard, or {@link SequenceNumbers#UNASSIGNED_SEQ_NO}
-     * @param onSyncRetentionLeases a callback when a new retention lease is created or an existing retention lease expires
+     * @param onAddRetentionLease a callback when a new retention lease is created or an existing retention lease expires
      */
     public ReplicationTracker(
             final ShardId shardId,
@@ -573,7 +573,7 @@ public ReplicationTracker(
             final long globalCheckpoint,
             final LongConsumer onGlobalCheckpointUpdated,
             final LongSupplier currentTimeMillisSupplier,
-            final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onSyncRetentionLeases) {
+            final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onAddRetentionLease) {
         super(shardId, indexSettings);
         assert globalCheckpoint >= SequenceNumbers.UNASSIGNED_SEQ_NO : "illegal initial global checkpoint: " + globalCheckpoint;
         this.shardAllocationId = allocationId;
@@ -585,7 +585,7 @@ public ReplicationTracker(
         checkpoints.put(allocationId, new CheckpointState(SequenceNumbers.UNASSIGNED_SEQ_NO, globalCheckpoint, false, false));
         this.onGlobalCheckpointUpdated = Objects.requireNonNull(onGlobalCheckpointUpdated);
         this.currentTimeMillisSupplier = Objects.requireNonNull(currentTimeMillisSupplier);
-        this.onSyncRetentionLeases = Objects.requireNonNull(onSyncRetentionLeases);
+        this.onAddRetentionLease = Objects.requireNonNull(onAddRetentionLease);
         this.pendingInSync = new HashSet<>();
         this.routingTable = null;
         this.replicationGroup = null;

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

@@ -1892,13 +1892,26 @@ public void addGlobalCheckpointListener(
     }
 
     /**
-     * Get all non-expired retention leases tracked on this shard.
+     * Get all retention leases tracked on this shard.
      *
      * @return the retention leases
      */
     public RetentionLeases getRetentionLeases() {
+        return getRetentionLeases(false).v2();
+    }
+
+    /**
+     * If the expire leases parameter is false, gets all retention leases tracked on this shard and otherwise first calculates
+     * expiration of existing retention leases, and then gets all non-expired retention leases tracked on this shard. Note that only the
+     * primary shard calculates which leases are expired, and if any have expired, syncs the retention leases to any replicas. If the
+     * expire leases parameter is true, this replication tracker must be in primary mode.
+     *
+     * @return a tuple indicating whether or not any retention leases were expired, and the non-expired retention leases
+     */
+    public Tuple<Boolean, RetentionLeases> getRetentionLeases(final boolean expireLeases) {
+        assert expireLeases == false || assertPrimaryMode();
         verifyNotClosed();
-        return replicationTracker.getRetentionLeases();
+        return replicationTracker.getRetentionLeases(expireLeases);
     }
 
     public RetentionLeaseStats getRetentionLeaseStats() {
@@ -1956,10 +1969,15 @@ public void updateRetentionLeasesOnReplica(final RetentionLeases retentionLeases
     /**
      * Syncs the current retention leases to all replicas.
      */
-    public void backgroundSyncRetentionLeases() {
+    public void syncRetentionLeases() {
         assert assertPrimaryMode();
         verifyNotClosed();
-        retentionLeaseSyncer.backgroundSync(shardId, getRetentionLeases());
+        final Tuple<Boolean, RetentionLeases> retentionLeases = getRetentionLeases(true);
+        if (retentionLeases.v1()) {
+            retentionLeaseSyncer.sync(shardId, retentionLeases.v2(), ActionListener.wrap(() -> {}));
+        } else {
+            retentionLeaseSyncer.backgroundSync(shardId, retentionLeases.v2());
+        }
     }
 
     /**