Refactoring checkpoint publishing to no longer be a Broadcast Action

This change adopts the same methodology as RetentionLeaseSyncer. A Transport client action is not required since we can skip the step of routing to the primary shard (the checkpoitn publisher is guaranteed to fire only from the primary). Now, checkpoint publishing directly invokes a TransportReplicationAction to perform the operation on the primary and replicas. PublishCheckpointAction has been reworked to be the TransportReplicationAction implementation rather than an ActionType. We leverage dependency injection to create the checkpoint publisher (and its internal action) at IndicesClusterStateService. This is plumed through to IndexShard which creates the refresh listener instance. All other transport layer classes tied to the original broadcast action are no longer required. Unrelated integration tests use a no-op/empty checkpoint publisher to satisfy their constructor/method argument. Signed-off-by: Kartik Ganesh <[email protected]>
opensearch-project · Mar 25, 2022 · 03bb12d · 03bb12d
1 parent 3b57a13
commit 03bb12d
Show file tree

Hide file tree

Showing 24 changed files with 250 additions and 325 deletions.
diff --git a/server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java b/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
         );
     }
 

diff --git a/server/src/main/java/org/opensearch/action/ActionModule.java b/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);

diff --git a/server/src/main/java/org/opensearch/client/IndicesAdminClient.java b/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).
      */

diff --git a/server/src/main/java/org/opensearch/client/support/AbstractClient.java b/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);

diff --git a/server/src/main/java/org/opensearch/index/IndexModule.java b/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;

diff --git a/server/src/main/java/org/opensearch/index/IndexService.java b/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);
         /*

diff --git a/server/src/main/java/org/opensearch/index/shard/CheckpointRefreshListener.java b/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);
         }
     }
 }
diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/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();

diff --git a/server/src/main/java/org/opensearch/indices/IndicesModule.java b/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();
     }
 
     /**

diff --git a/server/src/main/java/org/opensearch/indices/IndicesService.java b/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,

diff --git a/server/src/main/java/org/opensearch/indices/cluster/IndicesClusterStateService.java b/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,