POC for segment replication.

server/src/main/java/org/opensearch/action/ActionModule.java

@@ -270,6 +270,8 @@
 import org.opensearch.index.seqno.RetentionLeaseActions;
 import org.opensearch.indices.SystemIndices;
 import org.opensearch.indices.breaker.CircuitBreakerService;
+import org.opensearch.indices.replication.checkpoint.PublishCheckpointAction;
+import org.opensearch.indices.replication.checkpoint.TransportPublishCheckpointAction;
 import org.opensearch.persistent.CompletionPersistentTaskAction;
 import org.opensearch.persistent.RemovePersistentTaskAction;
 import org.opensearch.persistent.StartPersistentTaskAction;
@@ -588,6 +590,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(SimulateTemplateAction.INSTANCE, TransportSimulateTemplateAction.class);
         actions.register(ValidateQueryAction.INSTANCE, TransportValidateQueryAction.class);
         actions.register(RefreshAction.INSTANCE, TransportRefreshAction.class);
+        actions.register(PublishCheckpointAction.INSTANCE, TransportPublishCheckpointAction.class);
         actions.register(FlushAction.INSTANCE, TransportFlushAction.class);
         actions.register(SyncedFlushAction.INSTANCE, TransportSyncedFlushAction.class);
         actions.register(ForceMergeAction.INSTANCE, TransportForceMergeAction.class);

server/src/main/java/org/opensearch/action/admin/indices/refresh/RefreshResponse.java

@@ -61,11 +61,11 @@ public class RefreshResponse extends BroadcastResponse {
         declareBroadcastFields(PARSER);
     }
 
-    RefreshResponse(StreamInput in) throws IOException {
+    public RefreshResponse(StreamInput in) throws IOException {
         super(in);
     }
 
-    RefreshResponse(int totalShards, int successfulShards, int failedShards, List<DefaultShardOperationFailedException> shardFailures) {
+    public RefreshResponse(int totalShards, int successfulShards, int failedShards, List<DefaultShardOperationFailedException> shardFailures) {
         super(totalShards, successfulShards, failedShards, shardFailures);
     }
 

server/src/main/java/org/opensearch/client/IndicesAdminClient.java

@@ -135,6 +135,7 @@
 import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.cluster.metadata.IndexMetadata.APIBlock;
 import org.opensearch.common.Nullable;
+import org.opensearch.indices.replication.checkpoint.PublishCheckpointRequest;
 
 /**
  * Administrative actions/operations against indices.
@@ -405,6 +406,13 @@ public interface IndicesAdminClient extends OpenSearchClient {
      */
     void refresh(RefreshRequest request, ActionListener<RefreshResponse> listener);
 
+    /**
+     * Publish the latest primary checkpoint to replica shards.
+     * @param request {@link PublishCheckpointRequest} The PublishCheckpointRequest
+     * @param listener A listener to be notified with a result
+     */
+    void publishCheckpoint(PublishCheckpointRequest request, ActionListener<RefreshResponse> listener);
+
     /**
      * Explicitly refresh one or more indices (making the content indexed since the last refresh searchable).
      */

server/src/main/java/org/opensearch/client/support/AbstractClient.java

@@ -372,6 +372,8 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.common.xcontent.XContentType;
+import org.opensearch.indices.replication.checkpoint.PublishCheckpointAction;
+import org.opensearch.indices.replication.checkpoint.PublishCheckpointRequest;
 import org.opensearch.tasks.TaskId;
 import org.opensearch.threadpool.ThreadPool;
 
@@ -1632,6 +1634,11 @@ public void refresh(final RefreshRequest request, final ActionListener<RefreshRe
             execute(RefreshAction.INSTANCE, request, listener);
         }
 
+        @Override
+        public void publishCheckpoint(PublishCheckpointRequest request, ActionListener<RefreshResponse> listener) {
+            execute(PublishCheckpointAction.INSTANCE, request, listener);
+        }
+
         @Override
         public RefreshRequestBuilder prepareRefresh(String... indices) {
             return new RefreshRequestBuilder(this, RefreshAction.INSTANCE).setIndices(indices);

server/src/main/java/org/opensearch/index/IndexModule.java

@@ -75,6 +75,7 @@
 import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache;
 import org.opensearch.indices.mapper.MapperRegistry;
 import org.opensearch.indices.recovery.RecoveryState;
+import org.opensearch.indices.replication.checkpoint.TransportCheckpointPublisher;
 import org.opensearch.plugins.IndexStorePlugin;
 import org.opensearch.script.ScriptService;
 import org.opensearch.search.aggregations.support.ValuesSourceRegistry;
@@ -466,7 +467,8 @@ public IndexService newIndexService(
         IndicesFieldDataCache indicesFieldDataCache,
         NamedWriteableRegistry namedWriteableRegistry,
         BooleanSupplier idFieldDataEnabled,
-        ValuesSourceRegistry valuesSourceRegistry
+        ValuesSourceRegistry valuesSourceRegistry,
+        TransportCheckpointPublisher checkpointPublisher
     ) throws IOException {
         final IndexEventListener eventListener = freeze();
         Function<IndexService, CheckedFunction<DirectoryReader, DirectoryReader, IOException>> readerWrapperFactory = indexReaderWrapper
@@ -520,8 +522,8 @@ public IndexService newIndexService(
                 allowExpensiveQueries,
                 expressionResolver,
                 valuesSourceRegistry,
-                recoveryStateFactory
-            );
+                recoveryStateFactory,
+                checkpointPublisher);
             success = true;
             return indexService;
         } finally {

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

@@ -94,6 +94,7 @@
 import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache;
 import org.opensearch.indices.mapper.MapperRegistry;
 import org.opensearch.indices.recovery.RecoveryState;
+import org.opensearch.indices.replication.checkpoint.TransportCheckpointPublisher;
 import org.opensearch.plugins.IndexStorePlugin;
 import org.opensearch.script.ScriptService;
 import org.opensearch.search.aggregations.support.ValuesSourceRegistry;
@@ -165,6 +166,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final IndexNameExpressionResolver expressionResolver;
     private final Supplier<Sort> indexSortSupplier;
     private final ValuesSourceRegistry valuesSourceRegistry;
+    private final TransportCheckpointPublisher checkpointPublisher;
 
     public IndexService(
         IndexSettings indexSettings,
@@ -195,8 +197,8 @@ public IndexService(
         BooleanSupplier allowExpensiveQueries,
         IndexNameExpressionResolver expressionResolver,
         ValuesSourceRegistry valuesSourceRegistry,
-        IndexStorePlugin.RecoveryStateFactory recoveryStateFactory
-    ) {
+        IndexStorePlugin.RecoveryStateFactory recoveryStateFactory,
+        TransportCheckpointPublisher checkpointPublisher) {
         super(indexSettings);
         this.allowExpensiveQueries = allowExpensiveQueries;
         this.indexSettings = indexSettings;
@@ -206,6 +208,7 @@ public IndexService(
         this.circuitBreakerService = circuitBreakerService;
         this.expressionResolver = expressionResolver;
         this.valuesSourceRegistry = valuesSourceRegistry;
+        this.checkpointPublisher = checkpointPublisher;
         if (needsMapperService(indexSettings, indexCreationContext)) {
             assert indexAnalyzers != null;
             this.mapperService = new MapperService(
@@ -520,8 +523,8 @@ public synchronized IndexShard createShard(
                 indexingOperationListeners,
                 () -> globalCheckpointSyncer.accept(shardId),
                 retentionLeaseSyncer,
-                circuitBreakerService
-            );
+                circuitBreakerService,
+                checkpointPublisher);
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);
             shards = newMapBuilder(shards).put(shardId.id(), indexShard).immutableMap();
@@ -911,7 +914,9 @@ private void maybeRefreshEngine(boolean force) {
         if (indexSettings.getRefreshInterval().millis() > 0 || force) {
             for (IndexShard shard : this.shards.values()) {
                 try {
-                    shard.scheduledRefresh();
+                    if (shard.routingEntry().primary()) {
+                        shard.scheduledRefresh();
+                    }
                 } catch (IndexShardClosedException | AlreadyClosedException ex) {
                     // fine - continue;
                 }

server/src/main/java/org/opensearch/index/engine/Engine.java

@@ -249,6 +249,10 @@ public void verifyEngineBeforeIndexClosing() throws IllegalStateException {
         }
     }
 
+    public void updateCurrentInfos(byte[] infosBytes, long gen, long seqNo) throws IOException {};
+
+    public long getProcessedLocalCheckpoint() { return 0L; };
+
     /**
      * A throttling class that can be activated, causing the
      * {@code acquireThrottle} method to block on a lock when throttling
@@ -1196,6 +1200,20 @@ public abstract void forceMerge(
      */
     public abstract IndexCommitRef acquireLastIndexCommit(boolean flushFirst) throws EngineException;
 
+    /**
+     * Fetch a snapshot of the latest SegmentInfos from the engine and ensure that segment files are retained in the directory
+     * until closed.
+     * @return {@link SegmentInfosRef} - A ref to segmentInfos that must be closed for segment files to be deleted.
+     */
+    public SegmentInfosRef getLatestSegmentInfosSafe() { return null; };
+
+    /**
+     * Fetch a snapshot of the latest SegmentInfos from the engine.
+     * This method does not ensure that segment files are retained in the directory.
+     * @return {@link SegmentInfos}
+     */
+    public SegmentInfos getLatestSegmentInfos() { return null; };
+
     /**
      * Snapshots the most recent safe index commit from the engine.
      */
@@ -1999,6 +2017,28 @@ public IndexCommit getIndexCommit() {
         }
     }
 
+    public static class SegmentInfosRef implements Closeable {
+        private final AtomicBoolean closed = new AtomicBoolean();
+        private final CheckedRunnable<IOException> onClose;
+        private final SegmentInfos segmentInfos;
+
+        public SegmentInfosRef(SegmentInfos segmentInfos, CheckedRunnable<IOException> onClose) {
+            this.segmentInfos = segmentInfos;
+            this.onClose = onClose;
+        }
+
+        @Override
+        public void close() throws IOException {
+            if (closed.compareAndSet(false, true)) {
+                onClose.run();
+            }
+        }
+
+        public SegmentInfos getSegmentInfos() {
+            return segmentInfos;
+        }
+    }
+
     public void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue translogRetentionSize, long softDeletesRetentionOps) {
 
     }

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

@@ -95,6 +95,7 @@ public final class EngineConfig {
     private final CircuitBreakerService circuitBreakerService;
     private final LongSupplier globalCheckpointSupplier;
     private final Supplier<RetentionLeases> retentionLeasesSupplier;
+    private boolean isPrimary;
 
     /**
      * A supplier of the outstanding retention leases. This is used during merged operations to determine which operations that have been
@@ -169,8 +170,8 @@ public EngineConfig(
         LongSupplier globalCheckpointSupplier,
         Supplier<RetentionLeases> retentionLeasesSupplier,
         LongSupplier primaryTermSupplier,
-        TombstoneDocSupplier tombstoneDocSupplier
-    ) {
+        TombstoneDocSupplier tombstoneDocSupplier,
+        boolean isPrimary) {
         this(
             shardId,
             threadPool,
@@ -193,7 +194,7 @@ public EngineConfig(
             circuitBreakerService,
             globalCheckpointSupplier,
             retentionLeasesSupplier,
-            primaryTermSupplier,
+            isPrimary, primaryTermSupplier,
             tombstoneDocSupplier
         );
     }
@@ -223,7 +224,7 @@ public EngineConfig(
         CircuitBreakerService circuitBreakerService,
         LongSupplier globalCheckpointSupplier,
         Supplier<RetentionLeases> retentionLeasesSupplier,
-        LongSupplier primaryTermSupplier,
+        boolean isPrimary, LongSupplier primaryTermSupplier,
         TombstoneDocSupplier tombstoneDocSupplier
     ) {
         this.shardId = shardId;
@@ -237,6 +238,7 @@ public EngineConfig(
         this.codecService = codecService;
         this.eventListener = eventListener;
         codecName = indexSettings.getValue(INDEX_CODEC_SETTING);
+        this.isPrimary = isPrimary;
         // We need to make the indexing buffer for this shard at least as large
         // as the amount of memory that is available for all engines on the
         // local node so that decisions to flush segments to disk are made by
@@ -458,6 +460,10 @@ public LongSupplier getPrimaryTermSupplier() {
         return primaryTermSupplier;
     }
 
+    public boolean isPrimary() {
+        return isPrimary;
+    }
+
     /**
      * A supplier supplies tombstone documents which will be used in soft-update methods.
      * The returned document consists only _uid, _seqno, _term and _version fields; other metadata fields are excluded.

server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java

@@ -112,7 +112,8 @@ public EngineConfig newEngineConfig(
         LongSupplier globalCheckpointSupplier,
         Supplier<RetentionLeases> retentionLeasesSupplier,
         LongSupplier primaryTermSupplier,
-        EngineConfig.TombstoneDocSupplier tombstoneDocSupplier
+        EngineConfig.TombstoneDocSupplier tombstoneDocSupplier,
+        Boolean isPrimary
     ) {
 
         return new EngineConfig(
@@ -137,6 +138,7 @@ public EngineConfig newEngineConfig(
             circuitBreakerService,
             globalCheckpointSupplier,
             retentionLeasesSupplier,
+            isPrimary,
             primaryTermSupplier,
             tombstoneDocSupplier
         );