Refactoring checkpoint publishing to no longer be a Broadcast Action

server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java

@@ -84,7 +84,7 @@
 import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.indices.recovery.RecoveryState;
-import org.opensearch.indices.replication.checkpoint.TransportCheckpointPublisher;
+import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.search.builder.SearchSourceBuilder;
 import org.opensearch.test.DummyShardLock;
@@ -675,7 +675,7 @@ public final IndexShard newIndexShard(
             () -> {},
             RetentionLeaseSyncer.EMPTY,
             cbs,
-            new TransportCheckpointPublisher(client())
+            SegmentReplicationCheckpointPublisher.EMPTY
         );
     }
 

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

@@ -268,8 +268,6 @@
 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,7 +586,6 @@ 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(ForceMergeAction.INSTANCE, TransportForceMergeAction.class);
         actions.register(UpgradeAction.INSTANCE, TransportUpgradeAction.class);

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

@@ -132,7 +132,6 @@
 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.
@@ -403,13 +402,6 @@ 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

@@ -368,8 +368,6 @@
 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;
 
@@ -1610,11 +1608,6 @@ 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,7 +75,6 @@
 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;
@@ -467,8 +466,7 @@ public IndexService newIndexService(
         IndicesFieldDataCache indicesFieldDataCache,
         NamedWriteableRegistry namedWriteableRegistry,
         BooleanSupplier idFieldDataEnabled,
-        ValuesSourceRegistry valuesSourceRegistry,
-        TransportCheckpointPublisher checkpointPublisher
+        ValuesSourceRegistry valuesSourceRegistry
     ) throws IOException {
         final IndexEventListener eventListener = freeze();
         Function<IndexService, CheckedFunction<DirectoryReader, DirectoryReader, IOException>> readerWrapperFactory = indexReaderWrapper
@@ -522,8 +520,7 @@ public IndexService newIndexService(
                 allowExpensiveQueries,
                 expressionResolver,
                 valuesSourceRegistry,
-                recoveryStateFactory,
-                checkpointPublisher
+                recoveryStateFactory
             );
             success = true;
             return indexService;

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

@@ -94,7 +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.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.plugins.IndexStorePlugin;
 import org.opensearch.script.ScriptService;
 import org.opensearch.search.aggregations.support.ValuesSourceRegistry;
@@ -166,7 +166,6 @@ 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,
@@ -197,8 +196,7 @@ public IndexService(
         BooleanSupplier allowExpensiveQueries,
         IndexNameExpressionResolver expressionResolver,
         ValuesSourceRegistry valuesSourceRegistry,
-        IndexStorePlugin.RecoveryStateFactory recoveryStateFactory,
-        TransportCheckpointPublisher checkpointPublisher
+        IndexStorePlugin.RecoveryStateFactory recoveryStateFactory
     ) {
         super(indexSettings);
         this.allowExpensiveQueries = allowExpensiveQueries;
@@ -209,7 +207,6 @@ 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(
@@ -422,7 +419,8 @@ private long getAvgShardSizeInBytes() throws IOException {
     public synchronized IndexShard createShard(
         final ShardRouting routing,
         final Consumer<ShardId> globalCheckpointSyncer,
-        final RetentionLeaseSyncer retentionLeaseSyncer
+        final RetentionLeaseSyncer retentionLeaseSyncer,
+        final SegmentReplicationCheckpointPublisher checkpointPublisher
     ) throws IOException {
         Objects.requireNonNull(retentionLeaseSyncer);
         /*

server/src/main/java/org/opensearch/index/shard/CheckpointRefreshListener.java

@@ -11,7 +11,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.search.ReferenceManager;
-import org.opensearch.indices.replication.checkpoint.TransportCheckpointPublisher;
+import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 
 import java.io.IOException;
 
@@ -24,9 +24,9 @@ public class CheckpointRefreshListener implements ReferenceManager.RefreshListen
     protected static Logger logger = LogManager.getLogger(CheckpointRefreshListener.class);
 
     private final IndexShard shard;
-    private final TransportCheckpointPublisher publisher;
+    private final SegmentReplicationCheckpointPublisher publisher;
 
-    public CheckpointRefreshListener(IndexShard shard, TransportCheckpointPublisher publisher) {
+    public CheckpointRefreshListener(IndexShard shard, SegmentReplicationCheckpointPublisher publisher) {
         this.shard = shard;
         this.publisher = publisher;
     }
@@ -38,8 +38,8 @@ public void beforeRefresh() throws IOException {
 
     @Override
     public void afterRefresh(boolean didRefresh) throws IOException {
-        if (shard.routingEntry().primary()) {
-            publisher.publish(shard.getLatestReplicationCheckpoint());
+        if (didRefresh) {
+            publisher.publish(shard);
         }
     }
 }

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

@@ -161,7 +161,7 @@
 import org.opensearch.indices.recovery.RecoveryTarget;
 import org.opensearch.indices.replication.SegmentReplicationReplicaService;
 import org.opensearch.indices.replication.checkpoint.PublishCheckpointRequest;
-import org.opensearch.indices.replication.checkpoint.TransportCheckpointPublisher;
+import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.indices.replication.copy.PrimaryShardReplicationSource;
 import org.opensearch.indices.replication.copy.ReplicationCheckpoint;
 import org.opensearch.indices.replication.copy.ReplicationFailedException;
@@ -327,7 +327,7 @@ public IndexShard(
         final Runnable globalCheckpointSyncer,
         final RetentionLeaseSyncer retentionLeaseSyncer,
         final CircuitBreakerService circuitBreakerService,
-        final TransportCheckpointPublisher checkpointPublisher
+        final SegmentReplicationCheckpointPublisher checkpointPublisher
     ) throws IOException {
         super(shardRouting.shardId(), indexSettings);
         assert shardRouting.initializing();

server/src/main/java/org/opensearch/indices/IndicesModule.java

@@ -73,7 +73,7 @@
 import org.opensearch.index.shard.PrimaryReplicaSyncer;
 import org.opensearch.indices.cluster.IndicesClusterStateService;
 import org.opensearch.indices.mapper.MapperRegistry;
-import org.opensearch.indices.replication.checkpoint.TransportPublishShardCheckpointAction;
+import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.indices.store.IndicesStore;
 import org.opensearch.indices.store.TransportNodesListShardStoreMetadata;
 import org.opensearch.plugins.MapperPlugin;
@@ -276,8 +276,8 @@ protected void configure() {
         bind(PrimaryReplicaSyncer.class).asEagerSingleton();
         bind(RetentionLeaseSyncAction.class).asEagerSingleton();
         bind(RetentionLeaseBackgroundSyncAction.class).asEagerSingleton();
-        bind(TransportPublishShardCheckpointAction.class).asEagerSingleton();
         bind(RetentionLeaseSyncer.class).asEagerSingleton();
+        bind(SegmentReplicationCheckpointPublisher.class).asEagerSingleton();
     }
 
     /**

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -138,7 +138,7 @@
 import org.opensearch.indices.recovery.PeerRecoveryTargetService;
 import org.opensearch.indices.recovery.RecoveryState;
 import org.opensearch.indices.replication.SegmentReplicationReplicaService;
-import org.opensearch.indices.replication.checkpoint.TransportCheckpointPublisher;
+import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.indices.replication.copy.PrimaryShardReplicationSource;
 import org.opensearch.node.Node;
 import org.opensearch.plugins.IndexStorePlugin;
@@ -740,8 +740,7 @@ private synchronized IndexService createIndexService(
             indicesFieldDataCache,
             namedWriteableRegistry,
             this::isIdFieldDataEnabled,
-            valuesSourceRegistry,
-            new TransportCheckpointPublisher(client)
+            valuesSourceRegistry
         );
     }
 
@@ -840,6 +839,7 @@ public IndexShard createShard(
         final SegmentReplicationReplicaService segmentReplicationReplicaService,
         final SegmentReplicationReplicaService.SegmentReplicationListener segRepListener,
         final PrimaryShardReplicationSource replicationSource,
+        final SegmentReplicationCheckpointPublisher checkpointPublisher,
         final PeerRecoveryTargetService recoveryTargetService,
         final PeerRecoveryTargetService.RecoveryListener recoveryListener,
         final RepositoriesService repositoriesService,
@@ -850,11 +850,12 @@ public IndexShard createShard(
         final DiscoveryNode sourceNode
     ) throws IOException {
         Objects.requireNonNull(retentionLeaseSyncer);
+        Objects.requireNonNull(checkpointPublisher);
         ensureChangesAllowed();
         IndexService indexService = indexService(shardRouting.index());
         assert indexService != null;
         RecoveryState recoveryState = indexService.createRecoveryState(shardRouting, targetNode, sourceNode);
-        IndexShard indexShard = indexService.createShard(shardRouting, globalCheckpointSyncer, retentionLeaseSyncer);
+        IndexShard indexShard = indexService.createShard(shardRouting, globalCheckpointSyncer, retentionLeaseSyncer, checkpointPublisher);
         indexShard.addShardFailureCallback(onShardFailure);
         indexShard.startRecovery(
             recoveryState,

server/src/main/java/org/opensearch/indices/cluster/IndicesClusterStateService.java

@@ -81,6 +81,7 @@
 import org.opensearch.indices.recovery.RecoveryFailedException;
 import org.opensearch.indices.recovery.RecoveryState;
 import org.opensearch.indices.replication.SegmentReplicationReplicaService;
+import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.indices.replication.copy.PrimaryShardReplicationSource;
 import org.opensearch.indices.replication.copy.ReplicationFailedException;
 import org.opensearch.indices.replication.copy.SegmentReplicationState;
@@ -138,6 +139,7 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
     private final PrimaryReplicaSyncer primaryReplicaSyncer;
     private final Consumer<ShardId> globalCheckpointSyncer;
     private final RetentionLeaseSyncer retentionLeaseSyncer;
+    private final SegmentReplicationCheckpointPublisher checkpointPublisher;
 
     @Inject
     public IndicesClusterStateService(
@@ -156,7 +158,8 @@ public IndicesClusterStateService(
         final GlobalCheckpointSyncAction globalCheckpointSyncAction,
         final RetentionLeaseSyncer retentionLeaseSyncer,
         final SegmentReplicationReplicaService replicationReplicaService,
-        final PrimaryShardReplicationSource replicationSource
+        final PrimaryShardReplicationSource replicationSource,
+        final SegmentReplicationCheckpointPublisher checkpointPublisher
     ) {
         this(
             settings,
@@ -165,6 +168,7 @@ public IndicesClusterStateService(
             threadPool,
             replicationReplicaService,
             replicationSource,
+            checkpointPublisher,
             recoveryTargetService,
             shardStateAction,
             nodeMappingRefreshAction,
@@ -186,6 +190,7 @@ public IndicesClusterStateService(
         final ThreadPool threadPool,
         final SegmentReplicationReplicaService replicationReplicaService,
         final PrimaryShardReplicationSource replicationSource,
+        final SegmentReplicationCheckpointPublisher checkpointPublisher,
         final PeerRecoveryTargetService recoveryTargetService,
         final ShardStateAction shardStateAction,
         final NodeMappingRefreshAction nodeMappingRefreshAction,
@@ -200,6 +205,7 @@ public IndicesClusterStateService(
         this.settings = settings;
         this.segmentReplicationReplicaService = replicationReplicaService;
         this.replicationSource = replicationSource;
+        this.checkpointPublisher = checkpointPublisher;
         this.buildInIndexListener = Arrays.asList(peerRecoverySourceService, recoveryTargetService, searchService, snapshotShardsService);
         this.indicesService = indicesService;
         this.clusterService = clusterService;
@@ -632,6 +638,7 @@ private void createShard(DiscoveryNodes nodes, RoutingTable routingTable, ShardR
                 segmentReplicationReplicaService,
                 new ShardRoutingReplicationListener(shardRouting, primaryTerm),
                 replicationSource,
+                checkpointPublisher,
                 recoveryTargetService,
                 new ShardRoutingRecoveryListener(shardRouting, primaryTerm),
                 repositoriesService,
@@ -1030,6 +1037,7 @@ T createShard(
             SegmentReplicationReplicaService replicaService,
             SegmentReplicationReplicaService.SegmentReplicationListener segRepListener,
             PrimaryShardReplicationSource replicationSource,
+            SegmentReplicationCheckpointPublisher checkpointPublisher,
             PeerRecoveryTargetService recoveryTargetService,
             PeerRecoveryTargetService.RecoveryListener recoveryListener,
             RepositoriesService repositoriesService,