Introduce retention leases versioning

server/src/main/java/org/elasticsearch/index/engine/EngineConfig.java

@@ -35,15 +35,14 @@
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.codec.CodecService;
 import org.elasticsearch.index.mapper.ParsedDocument;
-import org.elasticsearch.index.seqno.RetentionLease;
+import org.elasticsearch.index.seqno.RetentionLeases;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.index.translog.TranslogConfig;
 import org.elasticsearch.indices.IndexingMemoryController;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.threadpool.ThreadPool;
 
-import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.LongSupplier;
@@ -81,15 +80,15 @@ public final class EngineConfig {
     @Nullable
     private final CircuitBreakerService circuitBreakerService;
     private final LongSupplier globalCheckpointSupplier;
-    private final Supplier<Collection<RetentionLease>> retentionLeasesSupplier;
+    private final Supplier<RetentionLeases> retentionLeasesSupplier;
 
     /**
      * A supplier of the outstanding retention leases. This is used during merged operations to determine which operations that have been
      * soft deleted should be retained.
      *
      * @return a supplier of outstanding retention leases
      */
-    public Supplier<Collection<RetentionLease>> retentionLeasesSupplier() {
+    public Supplier<RetentionLeases> retentionLeasesSupplier() {
         return retentionLeasesSupplier;
     }
 
@@ -141,7 +140,7 @@ public EngineConfig(ShardId shardId, String allocationId, ThreadPool threadPool,
                         List<ReferenceManager.RefreshListener> externalRefreshListener,
                         List<ReferenceManager.RefreshListener> internalRefreshListener, Sort indexSort,
                         CircuitBreakerService circuitBreakerService, LongSupplier globalCheckpointSupplier,
-                        Supplier<Collection<RetentionLease>> retentionLeasesSupplier,
+                        Supplier<RetentionLeases> retentionLeasesSupplier,
                         LongSupplier primaryTermSupplier,
                         TombstoneDocSupplier tombstoneDocSupplier) {
         this.shardId = shardId;

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -76,6 +76,7 @@
 import org.elasticsearch.index.merge.OnGoingMerge;
 import org.elasticsearch.index.seqno.LocalCheckpointTracker;
 import org.elasticsearch.index.seqno.RetentionLease;
+import org.elasticsearch.index.seqno.RetentionLeases;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.ElasticsearchMergePolicy;
@@ -2342,7 +2343,7 @@ protected void commitIndexWriter(final IndexWriter writer, final Translog transl
                      * We sample these from the policy (which occurs under a lock) to ensure that we have a consistent view of the minimum
                      * retained sequence number, and the retention leases.
                      */
-                    final Tuple<Long, Collection<RetentionLease>> retentionPolicy = softDeletesPolicy.getRetentionPolicy();
+                    final Tuple<Long, RetentionLeases> retentionPolicy = softDeletesPolicy.getRetentionPolicy();
                     commitData.put(Engine.MIN_RETAINED_SEQNO, Long.toString(retentionPolicy.v1()));
                     commitData.put(Engine.RETENTION_LEASES, RetentionLease.encodeRetentionLeases(retentionPolicy.v2()));
                 }

server/src/main/java/org/elasticsearch/index/engine/SoftDeletesPolicy.java

@@ -25,10 +25,10 @@
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.index.mapper.SeqNoFieldMapper;
 import org.elasticsearch.index.seqno.RetentionLease;
+import org.elasticsearch.index.seqno.RetentionLeases;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.translog.Translog;
 
-import java.util.Collection;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.LongSupplier;
@@ -46,15 +46,15 @@ final class SoftDeletesPolicy {
     private long retentionOperations;
     // The min seq_no value that is retained - ops after this seq# should exist in the Lucene index.
     private long minRetainedSeqNo;
-    private Collection<RetentionLease> retentionLeases;
+    private RetentionLeases retentionLeases;
     // provides the retention leases used to calculate the minimum sequence number to retain
-    private final Supplier<Collection<RetentionLease>> retentionLeasesSupplier;
+    private final Supplier<RetentionLeases> retentionLeasesSupplier;
 
     SoftDeletesPolicy(
             final LongSupplier globalCheckpointSupplier,
             final long minRetainedSeqNo,
             final long retentionOperations,
-            final Supplier<Collection<RetentionLease>> retentionLeasesSupplier) {
+            final Supplier<RetentionLeases> retentionLeasesSupplier) {
         this.globalCheckpointSupplier = globalCheckpointSupplier;
         this.retentionOperations = retentionOperations;
         this.minRetainedSeqNo = minRetainedSeqNo;
@@ -112,7 +112,7 @@ synchronized long getMinRetainedSeqNo() {
         return getRetentionPolicy().v1();
     }
 
-    public synchronized Tuple<Long, Collection<RetentionLease>> getRetentionPolicy() {
+    public synchronized Tuple<Long, RetentionLeases> getRetentionPolicy() {
         // do not advance if the retention lock is held
         if (retentionLockCount == 0) {
             /*
@@ -128,6 +128,7 @@ public synchronized Tuple<Long, Collection<RetentionLease>> getRetentionPolicy()
             // calculate the minimum sequence number to retain based on retention leases
             retentionLeases = retentionLeasesSupplier.get();
             final long minimumRetainingSequenceNumber = retentionLeases
+                    .retentionLeases()
                     .stream()
                     .mapToLong(RetentionLease::retainingSequenceNumber)
                     .min()

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

@@ -150,7 +150,7 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
      * 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.
      */
-    private final BiConsumer<Collection<RetentionLease>, ActionListener<ReplicationResponse>> onSyncRetentionLeases;
+    private final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onSyncRetentionLeases;
 
     /**
      * This set contains allocation IDs for which there is a thread actively waiting for the local checkpoint to advance to at least the
@@ -163,11 +163,12 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
      */
     volatile ReplicationGroup replicationGroup;
 
+    private volatile long version = 0;
     private final Map<String, RetentionLease> retentionLeases = new HashMap<>();
 
-    private Collection<RetentionLease> copyRetentionLeases() {
+    private RetentionLeases copyRetentionLeases() {
         assert Thread.holdsLock(this);
-        return Collections.unmodifiableCollection(new ArrayList<>(retentionLeases.values()));
+        return new RetentionLeases(version, Collections.unmodifiableCollection(new ArrayList<>(retentionLeases.values())));
     }
 
     /**
@@ -176,9 +177,9 @@ private Collection<RetentionLease> copyRetentionLeases() {
      *
      * @return the retention leases
      */
-    public Collection<RetentionLease> getRetentionLeases() {
+    public RetentionLeases getRetentionLeases() {
         final boolean wasPrimaryMode;
-        final Collection<RetentionLease> nonExpiredRetentionLeases;
+        final RetentionLeases nonExpiredRetentionLeases;
         synchronized (this) {
             if (primaryMode) {
                 // the primary calculates the non-expired retention leases and syncs them to replicas
@@ -197,6 +198,7 @@ public Collection<RetentionLease> getRetentionLeases() {
                 for (final RetentionLease expiredRetentionLease : expiredRetentionLeases) {
                     retentionLeases.remove(expiredRetentionLease.id());
                 }
+                version++;
             }
             /*
              * At this point, we were either in primary mode and have updated the non-expired retention leases into the tracking map, or
@@ -229,14 +231,15 @@ public RetentionLease addRetentionLease(
             final ActionListener<ReplicationResponse> listener) {
         Objects.requireNonNull(listener);
         final RetentionLease retentionLease;
-        final Collection<RetentionLease> currentRetentionLeases;
+        final RetentionLeases currentRetentionLeases;
         synchronized (this) {
             assert primaryMode;
             if (retentionLeases.containsKey(id)) {
                 throw new IllegalArgumentException("retention lease with ID [" + id + "] already exists");
             }
             retentionLease = new RetentionLease(id, retainingSequenceNumber, currentTimeMillisSupplier.getAsLong(), source);
             retentionLeases.put(id, retentionLease);
+            version++;
             currentRetentionLeases = copyRetentionLeases();
         }
         onSyncRetentionLeases.accept(currentRetentionLeases, listener);
@@ -260,6 +263,7 @@ public synchronized RetentionLease renewRetentionLease(final String id, final lo
         final RetentionLease retentionLease =
                 new RetentionLease(id, retainingSequenceNumber, currentTimeMillisSupplier.getAsLong(), source);
         final RetentionLease existingRetentionLease = retentionLeases.put(id, retentionLease);
+        version++;
         assert existingRetentionLease != null;
         assert existingRetentionLease.retainingSequenceNumber() <= retentionLease.retainingSequenceNumber() :
                 "retention lease renewal for [" + id + "]"
@@ -274,10 +278,14 @@ public synchronized RetentionLease renewRetentionLease(final String id, final lo
      *
      * @param retentionLeases the retention leases
      */
-    public synchronized void updateRetentionLeasesOnReplica(final Collection<RetentionLease> retentionLeases) {
+    public synchronized void updateRetentionLeasesOnReplica(final RetentionLeases retentionLeases) {
         assert primaryMode == false;
-        this.retentionLeases.clear();
-        this.retentionLeases.putAll(retentionLeases.stream().collect(Collectors.toMap(RetentionLease::id, Function.identity())));
+        if (retentionLeases.version() > version) {
+            this.retentionLeases.clear();
+            this.retentionLeases.putAll(
+                    retentionLeases.retentionLeases().stream().collect(Collectors.toMap(RetentionLease::id, Function.identity())));
+            this.version = retentionLeases.version();
+        }
     }
 
     public static class CheckpointState implements Writeable {
@@ -537,7 +545,7 @@ public ReplicationTracker(
             final long globalCheckpoint,
             final LongConsumer onGlobalCheckpointUpdated,
             final LongSupplier currentTimeMillisSupplier,
-            final BiConsumer<Collection<RetentionLease>, ActionListener<ReplicationResponse>> onSyncRetentionLeases) {
+            final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onSyncRetentionLeases) {
         super(shardId, indexSettings);
         assert globalCheckpoint >= SequenceNumbers.UNASSIGNED_SEQ_NO : "illegal initial global checkpoint: " + globalCheckpoint;
         this.shardAllocationId = allocationId;

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

@@ -173,9 +173,13 @@ static String encodeRetentionLease(final RetentionLease retentionLease) {
      * @param retentionLeases the retention leases
      * @return the encoding of the retention leases
      */
-    public static String encodeRetentionLeases(final Collection<RetentionLease> retentionLeases) {
+    public static String encodeRetentionLeases(final RetentionLeases retentionLeases) {
         Objects.requireNonNull(retentionLeases);
-        return retentionLeases.stream().map(RetentionLease::encodeRetentionLease).collect(Collectors.joining(","));
+        return String.format(
+                Locale.ROOT,
+                "version:%d;%s",
+                retentionLeases.version(),
+                retentionLeases.retentionLeases().stream().map(RetentionLease::encodeRetentionLease).collect(Collectors.joining(",")));
     }
 
     /**
@@ -200,20 +204,32 @@ static RetentionLease decodeRetentionLease(final String encodedRetentionLease) {
     }
 
     /**
-     * Decodes retention leases encoded by {@link #encodeRetentionLeases(Collection)}.
+     * Decodes retention leases encoded by {@link #encodeRetentionLeases(RetentionLeases)}.
      *
      * @param encodedRetentionLeases an encoded collection of retention leases
      * @return the decoded retention leases
      */
-    public static Collection<RetentionLease> decodeRetentionLeases(final String encodedRetentionLeases) {
+    public static RetentionLeases decodeRetentionLeases(final String encodedRetentionLeases) {
         Objects.requireNonNull(encodedRetentionLeases);
         if (encodedRetentionLeases.isEmpty()) {
-            return Collections.emptyList();
+            return RetentionLeases.EMPTY;
         }
-        assert Arrays.stream(encodedRetentionLeases.split(","))
-                .allMatch(s -> s.matches("id:[^:;,]+;retaining_seq_no:\\d+;timestamp:\\d+;source:[^:;,]+"))
-                : encodedRetentionLeases;
-        return Arrays.stream(encodedRetentionLeases.split(",")).map(RetentionLease::decodeRetentionLease).collect(Collectors.toList());
+        assert encodedRetentionLeases.matches("version:\\d+;.*") : encodedRetentionLeases;
+        final int firstSemicolon = encodedRetentionLeases.indexOf(";");
+        final long version = Long.parseLong(encodedRetentionLeases.substring("version:".length(), firstSemicolon));
+        final Collection<RetentionLease> retentionLeases;
+        if (firstSemicolon + 1 == encodedRetentionLeases.length()) {
+            retentionLeases = Collections.emptyList();
+        } else {
+            assert Arrays.stream(encodedRetentionLeases.substring(firstSemicolon + 1).split(","))
+                    .allMatch(s -> s.matches("id:[^:;,]+;retaining_seq_no:\\d+;timestamp:\\d+;source:[^:;,]+"))
+                    : encodedRetentionLeases;
+            retentionLeases = Arrays.stream(encodedRetentionLeases.substring(firstSemicolon + 1).split(","))
+                    .map(RetentionLease::decodeRetentionLease)
+                    .collect(Collectors.toList());
+        }
+
+        return new RetentionLeases(version, retentionLeases);
     }
 
     @Override

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

@@ -47,7 +47,6 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
-import java.util.Collection;
 import java.util.Objects;
 
 /**
@@ -99,7 +98,7 @@ public RetentionLeaseSyncAction(
      */
     public void syncRetentionLeasesForShard(
             final ShardId shardId,
-            final Collection<RetentionLease> retentionLeases,
+            final RetentionLeases retentionLeases,
             final ActionListener<ReplicationResponse> listener) {
         Objects.requireNonNull(shardId);
         Objects.requireNonNull(retentionLeases);
@@ -149,31 +148,31 @@ private void flush(final IndexShard indexShard) {
 
     public static final class Request extends ReplicatedWriteRequest<Request> {
 
-        private Collection<RetentionLease> retentionLeases;
+        private RetentionLeases retentionLeases;
 
-        public Collection<RetentionLease> getRetentionLeases() {
+        public RetentionLeases getRetentionLeases() {
             return retentionLeases;
         }
 
         public Request() {
 
         }
 
-        public Request(final ShardId shardId, final Collection<RetentionLease> retentionLeases) {
+        public Request(final ShardId shardId, final RetentionLeases retentionLeases) {
             super(Objects.requireNonNull(shardId));
             this.retentionLeases = Objects.requireNonNull(retentionLeases);
         }
 
         @Override
         public void readFrom(final StreamInput in) throws IOException {
             super.readFrom(in);
-            retentionLeases = in.readList(RetentionLease::new);
+            retentionLeases = new RetentionLeases(in);
         }
 
         @Override
         public void writeTo(final StreamOutput out) throws IOException {
             super.writeTo(Objects.requireNonNull(out));
-            out.writeCollection(retentionLeases);
+            retentionLeases.writeTo(out);
         }
 
         @Override

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

@@ -23,8 +23,6 @@
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.index.shard.ShardId;
 
-import java.util.Collection;
-
 /**
  * A functional interface that represents a method for syncing retention leases to replica shards after a new retention lease is added on
  * the primary.
@@ -42,7 +40,7 @@ public interface RetentionLeaseSyncer {
      */
     void syncRetentionLeasesForShard(
             ShardId shardId,
-            Collection<RetentionLease> retentionLeases,
+            RetentionLeases retentionLeases,
             ActionListener<ReplicationResponse> listener);
 
 }

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

@@ -0,0 +1,49 @@
+package org.elasticsearch.index.seqno;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Objects;
+
+public class RetentionLeases implements Writeable {
+
+    private final long version;
+
+    public long version() {
+        return version;
+    }
+
+    private final Collection<RetentionLease> retentionLeases;
+
+    public Collection<RetentionLease> retentionLeases() {
+        return retentionLeases;
+    }
+
+    public static RetentionLeases EMPTY = new RetentionLeases(0, Collections.emptyList());
+
+    public RetentionLeases(final long version, final Collection<RetentionLease> retentionLeases) {
+        if (version < 0) {
+            throw new IllegalArgumentException("version must be non-negative but was [" + version + "]");
+        }
+        Objects.requireNonNull(retentionLeases);
+        this.version = version;
+        this.retentionLeases = Collections.unmodifiableCollection(new ArrayList<>(retentionLeases));
+    }
+
+    public RetentionLeases(final StreamInput in) throws IOException {
+        version = in.readVLong();
+        retentionLeases = in.readList(RetentionLease::new);
+    }
+
+    @Override
+    public void writeTo(final StreamOutput out) throws IOException {
+        out.writeVLong(version);
+        out.writeCollection(retentionLeases);
+    }
+
+}