Update poller from cluster state applier to listener

Signed-off-by: Varun Bharadwaj <[email protected]>

Author: varunbharadwaj

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

@@ -729,7 +729,7 @@ public static final IndexShard newIndexShard(
             OpenSearchTestCase::randomBoolean,
             () -> indexService.getIndexSettings().getRefreshInterval(),
             indexService.getRefreshMutex(),
-            clusterService
+            clusterService.getClusterApplierService()
         );
     }
 

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

@@ -728,7 +728,7 @@ protected void closeInternal() {
                 fixedRefreshIntervalSchedulingEnabled,
                 this::getRefreshInterval,
                 refreshMutex,
-                clusterService
+                clusterService.getClusterApplierService()
             );
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);

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

@@ -41,7 +41,7 @@
 import org.apache.lucene.search.ReferenceManager;
 import org.apache.lucene.search.Sort;
 import org.apache.lucene.search.similarities.Similarity;
-import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.cluster.service.ClusterApplierService;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.settings.Setting;
@@ -114,7 +114,7 @@ public final class EngineConfig {
     private final BooleanSupplier startedPrimarySupplier;
     private final Comparator<LeafReader> leafSorter;
     private final Supplier<DocumentMapperForType> documentMapperForTypeSupplier;
-    private final ClusterService clusterService;
+    private final ClusterApplierService clusterApplierService;
 
     /**
      * A supplier of the outstanding retention leases. This is used during merged operations to determine which operations that have been
@@ -305,7 +305,7 @@ private EngineConfig(Builder builder) {
         this.leafSorter = builder.leafSorter;
         this.documentMapperForTypeSupplier = builder.documentMapperForTypeSupplier;
         this.indexReaderWarmer = builder.indexReaderWarmer;
-        this.clusterService = builder.clusterService;
+        this.clusterApplierService = builder.clusterApplierService;
     }
 
     /**
@@ -580,10 +580,10 @@ public Comparator<LeafReader> getLeafSorter() {
     }
 
     /**
-     * Returns the ClusterService instance.
+     * Returns the ClusterApplierService instance.
      */
-    public ClusterService getClusterService() {
-        return this.clusterService;
+    public ClusterApplierService getClusterApplierService() {
+        return this.clusterApplierService;
     }
 
     /**
@@ -621,7 +621,7 @@ public static class Builder {
         private Supplier<DocumentMapperForType> documentMapperForTypeSupplier;
         Comparator<LeafReader> leafSorter;
         private IndexWriter.IndexReaderWarmer indexReaderWarmer;
-        private ClusterService clusterService;
+        private ClusterApplierService clusterApplierService;
 
         public Builder shardId(ShardId shardId) {
             this.shardId = shardId;
@@ -768,8 +768,8 @@ public Builder indexReaderWarmer(IndexWriter.IndexReaderWarmer indexReaderWarmer
             return this;
         }
 
-        public Builder clusterService(ClusterService clusterService) {
-            this.clusterService = clusterService;
+        public Builder clusterApplierService(ClusterApplierService clusterApplierService) {
+            this.clusterApplierService = clusterApplierService;
             return this;
         }
 

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

@@ -18,7 +18,7 @@
 import org.apache.lucene.search.ReferenceManager;
 import org.apache.lucene.search.Sort;
 import org.apache.lucene.search.similarities.Similarity;
-import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.cluster.service.ClusterApplierService;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.index.shard.ShardId;
@@ -160,7 +160,7 @@ public EngineConfig newEngineConfig(
         Comparator<LeafReader> leafSorter,
         Supplier<DocumentMapperForType> documentMapperForTypeSupplier,
         IndexWriter.IndexReaderWarmer indexReaderWarmer,
-        ClusterService clusterService
+        ClusterApplierService clusterApplierService
     ) {
         CodecService codecServiceToUse = codecService;
         if (codecService == null && this.codecServiceFactory != null) {
@@ -196,7 +196,7 @@ public EngineConfig newEngineConfig(
             .leafSorter(leafSorter)
             .documentMapperForTypeSupplier(documentMapperForTypeSupplier)
             .indexReaderWarmer(indexReaderWarmer)
-            .clusterService(clusterService)
+            .clusterApplierService(clusterApplierService)
             .build();
     }
 

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

@@ -136,9 +136,9 @@ public void start() {
 
         // Register the poller with the ClusterService for receiving cluster state updates.
         // Also initialize cluster write block state in the poller.
-        if (engineConfig.getClusterService() != null) {
-            engineConfig.getClusterService().addStateApplier(streamPoller);
-            boolean isWriteBlockEnabled = engineConfig.getClusterService()
+        if (engineConfig.getClusterApplierService() != null) {
+            engineConfig.getClusterApplierService().addListener(streamPoller);
+            boolean isWriteBlockEnabled = engineConfig.getClusterApplierService()
                 .state()
                 .blocks()
                 .indexBlocked(ClusterBlockLevel.WRITE, engineConfig.getIndexSettings().getIndex().getName());

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

@@ -76,7 +76,7 @@
 import org.opensearch.cluster.routing.RecoverySource.SnapshotRecoverySource;
 import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.cluster.routing.ShardRoutingState;
-import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.cluster.service.ClusterApplierService;
 import org.opensearch.common.Booleans;
 import org.opensearch.common.CheckedConsumer;
 import org.opensearch.common.CheckedFunction;
@@ -376,7 +376,7 @@ Runnable getGlobalCheckpointSyncer() {
     private final Supplier<TimeValue> refreshInterval;
     private final Object refreshMutex;
     private volatile AsyncShardRefreshTask refreshTask;
-    private final ClusterService clusterService;
+    private final ClusterApplierService clusterApplierService;
 
     public IndexShard(
         final ShardRouting shardRouting,
@@ -414,7 +414,7 @@ public IndexShard(
         final Supplier<Boolean> fixedRefreshIntervalSchedulingEnabled,
         final Supplier<TimeValue> refreshInterval,
         final Object refreshMutex,
-        final ClusterService clusterService
+        final ClusterApplierService clusterApplierService
     ) throws IOException {
         super(shardRouting.shardId(), indexSettings);
         assert shardRouting.initializing();
@@ -521,7 +521,7 @@ public boolean shouldCache(Query query) {
         this.fixedRefreshIntervalSchedulingEnabled = fixedRefreshIntervalSchedulingEnabled;
         this.refreshInterval = refreshInterval;
         this.refreshMutex = Objects.requireNonNull(refreshMutex);
-        this.clusterService = clusterService;
+        this.clusterApplierService = clusterApplierService;
         synchronized (this.refreshMutex) {
             if (shardLevelRefreshEnabled) {
                 startRefreshTask();
@@ -4132,7 +4132,7 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) thro
             // timeseries
             () -> docMapper(),
             mergedSegmentWarmerFactory.get(this),
-            clusterService
+            clusterApplierService
         );
     }
 

server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java

@@ -441,8 +441,12 @@ public void setWriteBlockEnabled(boolean isWriteBlockEnabled) {
     }
 
     @Override
-    public void applyClusterState(final ClusterChangedEvent event) {
+    public void clusterChanged(ClusterChangedEvent event) {
         try {
+            if (event.blocksChanged() == false) {
+                return;
+            }
+
             final ClusterState state = event.state();
             isWriteBlockEnabled = state.blocks().indexBlocked(ClusterBlockLevel.WRITE, indexName);
 

server/src/main/java/org/opensearch/indices/pollingingest/StreamPoller.java

@@ -8,7 +8,7 @@
 
 package org.opensearch.indices.pollingingest;
 
-import org.opensearch.cluster.ClusterStateApplier;
+import org.opensearch.cluster.ClusterStateListener;
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.index.IngestionShardPointer;
 
@@ -17,7 +17,7 @@
 /**
  * A poller for reading messages from an ingestion shard. This is used in the ingestion engine.
  */
-public interface StreamPoller extends Closeable, ClusterStateApplier {
+public interface StreamPoller extends Closeable, ClusterStateListener {
 
     String BATCH_START = "batch_start";
 

server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java

@@ -504,13 +504,13 @@ public void testClusterStateChange() {
             .build();
 
         ClusterChangedEvent event1 = new ClusterChangedEvent("test", state2, state1);
-        poller.applyClusterState(event1);
+        poller.clusterChanged(event1);
         assertTrue(poller.isWriteBlockEnabled());
 
         // remove write block
         ClusterState state3 = ClusterState.builder(ClusterName.DEFAULT).build();
         ClusterChangedEvent event2 = new ClusterChangedEvent("test", state3, state2);
-        poller.applyClusterState(event2);
+        poller.clusterChanged(event2);
         assertFalse(poller.isWriteBlockEnabled());
 
     }

test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java

@@ -729,7 +729,7 @@ protected IndexShard newShard(
                 () -> Boolean.FALSE,
                 indexSettings::getRefreshInterval,
                 new Object(),
-                clusterService
+                clusterService.getClusterApplierService()
             );
             indexShard.addShardFailureCallback(DEFAULT_SHARD_FAILURE_HANDLER);
             if (remoteStoreStatsTrackerFactory != null) {