Introduce global checkpoint background sync

core/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java

@@ -173,6 +173,7 @@ public void onResponse(ReplicaResponse response) {
                 successfulShards.incrementAndGet();
                 try {
                     primary.updateLocalCheckpointForShard(shard.allocationId().getId(), response.localCheckpoint());
+                    primary.updateGlobalCheckpointForShard(shard.allocationId().getId(), globalCheckpoint);
                 } catch (final AlreadyClosedException e) {
                     // okay, the index was deleted or this shard was never activated after a relocation; fall through and finish normally
                 } catch (final Exception e) {
@@ -315,6 +316,14 @@ public interface Primary<
          */
         void updateLocalCheckpointForShard(String allocationId, long checkpoint);
 
+        /**
+         * Update the local knowledge of the global checkpoint for the specified allocation ID.
+         *
+         * @param allocationId     the allocation ID to update the global checkpoint for
+         * @param globalCheckpoint the global checkpoint
+         */
+        void updateGlobalCheckpointForShard(String allocationId, long globalCheckpoint);
+
         /**
          * Returns the local checkpoint on the primary shard.
          *

core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -1006,6 +1006,11 @@ public void updateLocalCheckpointForShard(String allocationId, long checkpoint)
             indexShard.updateLocalCheckpointForShard(allocationId, checkpoint);
         }
 
+        @Override
+        public void updateGlobalCheckpointForShard(final String allocationId, final long globalCheckpoint) {
+            indexShard.updateGlobalCheckpointForShard(allocationId, globalCheckpoint);
+        }
+
         @Override
         public long localCheckpoint() {
             return indexShard.getLocalCheckpoint();

core/src/main/java/org/elasticsearch/cluster/metadata/MetaDataCreateIndexService.java

@@ -428,7 +428,7 @@ public ClusterState execute(ClusterState currentState) throws Exception {
                         (tmpImd.getNumberOfReplicas() + 1) + "]");
                 }
                 // create the index here (on the master) to validate it can be created, as well as adding the mapping
-                final IndexService indexService = indicesService.createIndex(tmpImd, Collections.emptyList());
+                final IndexService indexService = indicesService.createIndex(tmpImd, Collections.emptyList(), s -> {});
                 createdIndex = indexService.index();
                 // now add the mappings
                 MapperService mapperService = indexService.mapperService();

core/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexAliasesService.java

@@ -139,7 +139,7 @@ ClusterState innerExecute(ClusterState currentState, Iterable<AliasAction> actio
                             if (indexService == null) {
                                 // temporarily create the index and add mappings so we can parse the filter
                                 try {
-                                    indexService = indicesService.createIndex(index, emptyList());
+                                    indexService = indicesService.createIndex(index, emptyList(), s -> {});
                                     indicesToClose.add(index.getIndex());
                                 } catch (IOException e) {
                                     throw new ElasticsearchException("Failed to create temporary index for parsing the alias", e);

core/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexTemplateService.java

@@ -213,7 +213,7 @@ private static void validateAndAddTemplate(final PutRequest request, IndexTempla
                 .build();
 
             final IndexMetaData tmpIndexMetadata = IndexMetaData.builder(temporaryIndexName).settings(dummySettings).build();
-            IndexService dummyIndexService = indicesService.createIndex(tmpIndexMetadata, Collections.emptyList());
+            IndexService dummyIndexService = indicesService.createIndex(tmpIndexMetadata, Collections.emptyList(), s -> {});
             createdIndex = dummyIndexService.index();
 
             templateBuilder.order(request.order);

core/src/main/java/org/elasticsearch/cluster/metadata/MetaDataMappingService.java

@@ -145,7 +145,7 @@ ClusterState executeRefresh(final ClusterState currentState, final List<RefreshT
             IndexService indexService = indicesService.indexService(indexMetaData.getIndex());
             if (indexService == null) {
                 // we need to create the index here, and add the current mapping to it, so we can merge
-                indexService = indicesService.createIndex(indexMetaData, Collections.emptyList());
+                indexService = indicesService.createIndex(indexMetaData, Collections.emptyList(), s -> {});
                 removeIndex = true;
                 indexService.mapperService().merge(indexMetaData, MergeReason.MAPPING_RECOVERY, true);
             }

core/src/main/java/org/elasticsearch/index/IndexModule.java

@@ -40,6 +40,7 @@
 import org.elasticsearch.index.shard.IndexSearcherWrapper;
 import org.elasticsearch.index.shard.IndexingOperationListener;
 import org.elasticsearch.index.shard.SearchOperationListener;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.similarity.BM25SimilarityProvider;
 import org.elasticsearch.index.similarity.SimilarityProvider;
 import org.elasticsearch.index.similarity.SimilarityService;
@@ -330,7 +331,8 @@ public IndexService newIndexService(
             IndicesQueryCache indicesQueryCache,
             MapperRegistry mapperRegistry,
             IndicesFieldDataCache indicesFieldDataCache,
-            NamedWriteableRegistry namedWriteableRegistry)
+            NamedWriteableRegistry namedWriteableRegistry,
+            Consumer<ShardId> globalCheckpointSyncer)
         throws IOException {
         final IndexEventListener eventListener = freeze();
         IndexSearcherWrapperFactory searcherWrapperFactory = indexSearcherWrapper.get() == null
@@ -365,7 +367,7 @@ public IndexService newIndexService(
                 new SimilarityService(indexSettings, scriptService, similarities),
                 shardStoreDeleter, analysisRegistry, engineFactory.get(), circuitBreakerService, bigArrays, threadPool, scriptService,
                 client, queryCache, store, eventListener, searcherWrapperFactory, mapperRegistry,
-                indicesFieldDataCache, searchOperationListeners, indexOperationListeners, namedWriteableRegistry);
+                indicesFieldDataCache, searchOperationListeners, indexOperationListeners, namedWriteableRegistry, globalCheckpointSyncer);
     }
 
     /**

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

@@ -82,6 +82,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
@@ -109,10 +110,12 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final AtomicBoolean deleted = new AtomicBoolean(false);
     private final IndexSettings indexSettings;
-    private final List<IndexingOperationListener> indexingOperationListeners;
     private final List<SearchOperationListener> searchOperationListeners;
+    private final List<IndexingOperationListener> indexingOperationListeners;
+    private final Consumer<ShardId> globalCheckpointSyncer;
     private volatile AsyncRefreshTask refreshTask;
     private volatile AsyncTranslogFSync fsyncTask;
+    private volatile AsyncGlobalCheckpointTask globalCheckpointTask;
 
     // 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";
@@ -145,7 +148,8 @@ public IndexService(
             IndicesFieldDataCache indicesFieldDataCache,
             List<SearchOperationListener> searchOperationListeners,
             List<IndexingOperationListener> indexingOperationListeners,
-            NamedWriteableRegistry namedWriteableRegistry) throws IOException {
+            NamedWriteableRegistry namedWriteableRegistry,
+            Consumer<ShardId> globalCheckpointSyncer) throws IOException {
         super(indexSettings);
         this.indexSettings = indexSettings;
         this.xContentRegistry = xContentRegistry;
@@ -182,11 +186,13 @@ public IndexService(
         this.engineFactory = engineFactory;
         // initialize this last -- otherwise if the wrapper requires any other member to be non-null we fail with an NPE
         this.searcherWrapper = wrapperFactory.newWrapper(this);
-        this.indexingOperationListeners = Collections.unmodifiableList(indexingOperationListeners);
         this.searchOperationListeners = Collections.unmodifiableList(searchOperationListeners);
+        this.indexingOperationListeners = Collections.unmodifiableList(indexingOperationListeners);
+        this.globalCheckpointSyncer = globalCheckpointSyncer;
         // kick off async ops for the first shard in this index
         this.refreshTask = new AsyncRefreshTask(this);
         this.trimTranslogTask = new AsyncTrimTranslogTask(this);
+        this.globalCheckpointTask = new AsyncGlobalCheckpointTask(this);
         rescheduleFsyncTask(indexSettings.getTranslogDurability());
     }
 
@@ -268,7 +274,15 @@ public synchronized void close(final String reason, boolean delete) throws IOExc
                     }
                 }
             } finally {
-                IOUtils.close(bitsetFilterCache, indexCache, indexFieldData, mapperService, refreshTask, fsyncTask, trimTranslogTask);
+                IOUtils.close(
+                        bitsetFilterCache,
+                        indexCache,
+                        indexFieldData,
+                        mapperService,
+                        refreshTask,
+                        fsyncTask,
+                        trimTranslogTask,
+                        globalCheckpointTask);
             }
         }
     }
@@ -365,7 +379,7 @@ public synchronized IndexShard createShard(ShardRouting routing) throws IOExcept
             indexShard = new IndexShard(routing, this.indexSettings, path, store, indexSortSupplier,
                 indexCache, mapperService, similarityService, engineFactory,
                 eventListener, searcherWrapper, threadPool, bigArrays, engineWarmer,
-                    searchOperationListeners, indexingOperationListeners);
+                    searchOperationListeners, indexingOperationListeners, () -> globalCheckpointSyncer.accept(shardId));
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);
             shards = newMapBuilder(shards).put(shardId.id(), indexShard).immutableMap();
@@ -710,6 +724,30 @@ private void maybeTrimTranslog() {
         }
     }
 
+    private void syncGlobalCheckpoints() {
+        for (final IndexShard shard : this.shards.values()) {
+            if (shard.routingEntry().active() && shard.routingEntry().primary()) {
+                switch (shard.state()) {
+                    case CREATED:
+                    case RECOVERING:
+                    case CLOSED:
+                        continue;
+                    case POST_RECOVERY:
+                    case STARTED:
+                    case RELOCATED:
+                        try {
+                            shard.maybeSyncGlobalCheckpoint();
+                        } catch (final AlreadyClosedException | IndexShardClosedException e) {
+                            // the shard was closed concurrently, continue
+                        }
+                        continue;
+                    default:
+                        throw new IllegalStateException("unknown state [" + shard.state() + "]");
+                }
+            }
+        }
+    }
+
     abstract static class BaseAsyncTask implements Runnable, Closeable {
         protected final IndexService indexService;
         protected final ThreadPool threadPool;
@@ -877,6 +915,39 @@ public String toString() {
         }
     }
 
+    private static final TimeValue GLOBAL_CHECKPOINT_SYNC_INTERVAL = TimeValue.timeValueSeconds(30);
+
+    /**
+     * Background task that syncs the global checkpoint to replicas.
+     */
+    final class AsyncGlobalCheckpointTask extends BaseAsyncTask {
+
+        AsyncGlobalCheckpointTask(final IndexService indexService) {
+            // index.global_checkpoint_sync_interval is not a real setting, it is only registered in tests
+            super(
+                    indexService,
+                    indexService
+                            .getIndexSettings()
+                            .getSettings()
+                            .getAsTime("index.global_checkpoint_sync.interval", GLOBAL_CHECKPOINT_SYNC_INTERVAL));
+        }
+
+        @Override
+        protected void runInternal() {
+            indexService.syncGlobalCheckpoints();
+        }
+
+        @Override
+        protected String getThreadPool() {
+            return ThreadPool.Names.GENERIC;
+        }
+
+        @Override
+        public String toString() {
+            return "global_checkpoint_sync";
+        }
+    }
+
     AsyncRefreshTask getRefreshTask() { // for tests
         return refreshTask;
     }
@@ -885,6 +956,10 @@ AsyncTranslogFSync getFsyncTask() { // for tests
         return fsyncTask;
     }
 
+    AsyncGlobalCheckpointTask getGlobalCheckpointTask() {
+        return globalCheckpointTask;
+    }
+
     /**
      * Clears the caches for the given shard id if the shard is still allocated on this node
      */

core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java

@@ -47,7 +47,7 @@
 public class GlobalCheckpointSyncAction extends TransportReplicationAction<
         GlobalCheckpointSyncAction.Request,
         GlobalCheckpointSyncAction.Request,
-        ReplicationResponse> implements IndexEventListener {
+        ReplicationResponse> {
 
     public static String ACTION_NAME = "indices:admin/seq_no/global_checkpoint_sync";
 
@@ -76,6 +76,10 @@ public GlobalCheckpointSyncAction(
                 ThreadPool.Names.SAME);
     }
 
+    public void updateGlobalCheckpointForShard(final ShardId shardId) {
+        execute(new Request(shardId));
+    }
+
     @Override
     protected ReplicationResponse newResponseInstance() {
         return new ReplicationResponse();
@@ -93,11 +97,6 @@ protected void sendReplicaRequest(
         }
     }
 
-    @Override
-    public void onShardInactive(final IndexShard indexShard) {
-        execute(new Request(indexShard.shardId()));
-    }
-
     @Override
     protected PrimaryResult<Request, ReplicationResponse> shardOperationOnPrimary(
             final Request request, final IndexShard indexShard) throws Exception {

core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointTracker.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.index.seqno;
 
+import com.carrotsearch.hppc.ObjectLongHashMap;
+import com.carrotsearch.hppc.ObjectLongMap;
 import org.elasticsearch.cluster.routing.AllocationId;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -496,6 +498,38 @@ public synchronized void updateLocalCheckpoint(final String allocationId, final
         assert invariant();
     }
 
+    private ObjectLongMap<String> globalCheckpoints = new ObjectLongHashMap<>();
+
+    /**
+     * Update the local knowledge of the global checkpoint for the specified allocation ID.
+     *
+     * @param allocationId     the allocation ID to update the global checkpoint for
+     * @param globalCheckpoint the global checkpoint
+     */
+    synchronized void updateGlobalCheckpointForShard(final String allocationId, final long globalCheckpoint) {
+        assert primaryMode;
+        assert handoffInProgress == false;
+        final long current = globalCheckpoints.getOrDefault(allocationId, SequenceNumbers.UNASSIGNED_SEQ_NO);
+        if (globalCheckpoint > current) {
+            globalCheckpoints.put(allocationId, globalCheckpoint);
+        }
+    }
+
+    /**
+     * Returns the global checkpoints for all shards.
+     *
+     * @param allocationId the allocationId to use for the global checkpoint on the primary
+     *
+     * @return the global checkpoints for all shards
+     */
+    synchronized ObjectLongMap<String> getGlobalCheckpoints(final String allocationId) {
+        assert primaryMode;
+        assert handoffInProgress == false;
+        final ObjectLongMap<String> copy = new ObjectLongHashMap<>(globalCheckpoints);
+        copy.put(allocationId, globalCheckpoint);
+        return copy;
+    }
+
     /**
      * Computes the global checkpoint based on the given local checkpoints. In case where there are entries preventing the
      * computation to happen (for example due to blocking), it returns the fallback value.
@@ -585,6 +619,7 @@ public synchronized void completeRelocationHandoff() {
                 lcps.localCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
             }
         });
+        globalCheckpoints.clear();
         assert invariant();
     }
 

core/src/main/java/org/elasticsearch/index/seqno/SequenceNumbersService.java

@@ -19,7 +19,9 @@
 
 package org.elasticsearch.index.seqno;
 
+import com.carrotsearch.hppc.ObjectLongMap;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.shard.ReplicationGroup;
@@ -131,6 +133,16 @@ public void updateLocalCheckpointForShard(final String allocationId, final long
         globalCheckpointTracker.updateLocalCheckpoint(allocationId, checkpoint);
     }
 
+    /**
+     * Update the local knowledge of the global checkpoint for the specified allocation ID.
+     *
+     * @param allocationId     the allocation ID to update the global checkpoint for
+     * @param globalCheckpoint the global checkpoint
+     */
+    public void updateGlobalCheckpointForShard(final String allocationId, final long globalCheckpoint) {
+        globalCheckpointTracker.updateGlobalCheckpointForShard(allocationId, globalCheckpoint);
+    }
+
     /**
      * Called when the recovery process for a shard is ready to open the engine on the target shard.
      * See {@link GlobalCheckpointTracker#initiateTracking(String)} for details.
@@ -179,6 +191,17 @@ public long getGlobalCheckpoint() {
         return globalCheckpointTracker.getGlobalCheckpoint();
     }
 
+    /**
+     * Returns the global checkpoints for all shards.
+     *
+     * @param allocationId the allocationId to use for the global checkpoint on the primary
+     *
+     * @return the global checkpoints for all shards
+     */
+    public ObjectLongMap<String> getGlobalCheckpoints(final String allocationId) {
+        return globalCheckpointTracker.getGlobalCheckpoints(allocationId);
+    }
+
     /**
      * Updates the global checkpoint on a replica shard after it has been updated by the primary.
      *