Inject extension into service

Author: n1v0lg

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

@@ -215,6 +215,8 @@ public TransportSearchAction(
         this.searchResponseMetrics = searchResponseMetrics;
         this.client = client;
         this.usageService = usageService;
+        // just showing injection
+        logger.info("Refresh remote connections: {}", searchService.forceRefreshRemoteConnections());
     }
 
     private Map<String, OriginalIndices> buildPerIndexOriginalIndices(

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -23,7 +23,6 @@
 import org.elasticsearch.action.bulk.IncrementalBulkService;
 import org.elasticsearch.action.datastreams.autosharding.DataStreamAutoShardingService;
 import org.elasticsearch.action.ingest.ReservedPipelineAction;
-import org.elasticsearch.search.internal.CrossClusterSearchExtension;
 import org.elasticsearch.action.search.OnlinePrewarmingService;
 import org.elasticsearch.action.search.OnlinePrewarmingServiceProvider;
 import org.elasticsearch.action.search.SearchExecutionStatsCollector;
@@ -204,6 +203,7 @@
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.SearchUtils;
 import org.elasticsearch.search.aggregations.support.AggregationUsageService;
+import org.elasticsearch.search.internal.CrossClusterSearchExtension;
 import org.elasticsearch.shutdown.PluginShutdownService;
 import org.elasticsearch.snapshots.InternalSnapshotsInfoService;
 import org.elasticsearch.snapshots.RepositoryIntegrityHealthIndicatorService;
@@ -1021,7 +1021,7 @@ public Map<String, String> queryFields() {
             CrossClusterSearchExtension.Default::new
         );
 
-        logger.info("Cross-cluster search force reconnect: [{}]", crossClusterSearchExtension.forceReconnectBehaviorSupplier().get());
+        logger.info("Cross-cluster search force reconnect: [{}]", crossClusterSearchExtension.forceRefreshRemoteConnections().get());
 
         ActionModule actionModule = new ActionModule(
             settings,
@@ -1213,7 +1213,8 @@ public Map<String, String> queryFields() {
             circuitBreakerService,
             systemIndices.getExecutorSelector(),
             telemetryProvider.getTracer(),
-            onlinePrewarmingService
+            onlinePrewarmingService,
+            crossClusterSearchExtension
         );
 
         final ShutdownPrepareService shutdownPrepareService = new ShutdownPrepareService(settings, httpServerTransport, terminationHandler);

server/src/main/java/org/elasticsearch/node/NodeServiceProvider.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.fetch.FetchPhase;
+import org.elasticsearch.search.internal.CrossClusterSearchExtension;
 import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.telemetry.tracing.Tracer;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -135,7 +136,8 @@ SearchService newSearchService(
         CircuitBreakerService circuitBreakerService,
         ExecutorSelector executorSelector,
         Tracer tracer,
-        OnlinePrewarmingService onlinePrewarmingService
+        OnlinePrewarmingService onlinePrewarmingService,
+        CrossClusterSearchExtension crossClusterSearchExtension
     ) {
         return new SearchService(
             clusterService,
@@ -147,7 +149,8 @@ SearchService newSearchService(
             circuitBreakerService,
             executorSelector,
             tracer,
-            onlinePrewarmingService
+            onlinePrewarmingService,
+            crossClusterSearchExtension
         );
     }
 

server/src/main/java/org/elasticsearch/search/SearchService.java

@@ -104,6 +104,7 @@
 import org.elasticsearch.search.fetch.subphase.ScriptFieldsContext.ScriptField;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
 import org.elasticsearch.search.internal.AliasFilter;
+import org.elasticsearch.search.internal.CrossClusterSearchExtension;
 import org.elasticsearch.search.internal.InternalScrollSearchRequest;
 import org.elasticsearch.search.internal.LegacyReaderContext;
 import org.elasticsearch.search.internal.ReaderContext;
@@ -368,6 +369,8 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
 
     private final Tracer tracer;
 
+    private final CrossClusterSearchExtension crossClusterSearchExtension;
+
     public SearchService(
         ClusterService clusterService,
         IndicesService indicesService,
@@ -378,7 +381,8 @@ public SearchService(
         CircuitBreakerService circuitBreakerService,
         ExecutorSelector executorSelector,
         Tracer tracer,
-        OnlinePrewarmingService onlinePrewarmingService
+        OnlinePrewarmingService onlinePrewarmingService,
+        CrossClusterSearchExtension crossClusterSearchExtension
     ) {
         Settings settings = clusterService.getSettings();
         this.threadPool = threadPool;
@@ -444,6 +448,8 @@ public SearchService(
         clusterService.getClusterSettings()
             .addSettingsUpdateConsumer(MEMORY_ACCOUNTING_BUFFER_SIZE, newValue -> this.memoryAccountingBufferSize = newValue.getBytes());
         prewarmingMaxPoolFactorThreshold = PREWARMING_THRESHOLD_THREADPOOL_SIZE_FACTOR_POOL_SIZE.get(settings);
+
+        this.crossClusterSearchExtension = crossClusterSearchExtension;
     }
 
     public CircuitBreaker getCircuitBreaker() {
@@ -458,6 +464,10 @@ private void setEnableSearchWorkerThreads(boolean enableSearchWorkerThreads) {
         }
     }
 
+    public boolean forceRefreshRemoteConnections() {
+        return crossClusterSearchExtension.forceRefreshRemoteConnections().get();
+    }
+
     private void setEnableQueryPhaseParallelCollection(boolean enableQueryPhaseParallelCollection) {
         this.enableQueryPhaseParallelCollection = enableQueryPhaseParallelCollection;
     }

server/src/main/java/org/elasticsearch/search/internal/CrossClusterSearchExtension.java

@@ -13,13 +13,13 @@
 
 public interface CrossClusterSearchExtension {
 
-    Supplier<Boolean> forceReconnectBehaviorSupplier();
+    Supplier<Boolean> forceRefreshRemoteConnections();
 
     class Default implements CrossClusterSearchExtension {
         public Default() {}
 
         @Override
-        public Supplier<Boolean> forceReconnectBehaviorSupplier() {
+        public Supplier<Boolean> forceRefreshRemoteConnections() {
             return () -> false;
         }
     }

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -186,6 +186,7 @@
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.fetch.FetchPhase;
+import org.elasticsearch.search.internal.CrossClusterSearchExtension;
 import org.elasticsearch.telemetry.TelemetryProvider;
 import org.elasticsearch.telemetry.tracing.Tracer;
 import org.elasticsearch.test.ClusterServiceUtils;
@@ -2557,7 +2558,8 @@ public RecyclerBytesStreamOutput newNetworkBytesStream() {
                     new NoneCircuitBreakerService(),
                     EmptySystemIndices.INSTANCE.getExecutorSelector(),
                     Tracer.NOOP,
-                    OnlinePrewarmingService.NOOP
+                    OnlinePrewarmingService.NOOP,
+                    new CrossClusterSearchExtension.Default()
                 );
 
                 final SnapshotFilesProvider snapshotFilesProvider = new SnapshotFilesProvider(repositoriesService);

test/framework/src/main/java/org/elasticsearch/node/MockNode.java

@@ -42,6 +42,7 @@
 import org.elasticsearch.search.MockSearchService;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.fetch.FetchPhase;
+import org.elasticsearch.search.internal.CrossClusterSearchExtension;
 import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.telemetry.tracing.Tracer;
 import org.elasticsearch.test.ESTestCase;
@@ -103,7 +104,8 @@ SearchService newSearchService(
             CircuitBreakerService circuitBreakerService,
             ExecutorSelector executorSelector,
             Tracer tracer,
-            OnlinePrewarmingService onlinePrewarmingService
+            OnlinePrewarmingService onlinePrewarmingService,
+            CrossClusterSearchExtension crossClusterSearchExtension
         ) {
             if (pluginsService.filterPlugins(MockSearchService.TestPlugin.class).findAny().isEmpty()) {
                 return super.newSearchService(
@@ -117,7 +119,8 @@ SearchService newSearchService(
                     circuitBreakerService,
                     executorSelector,
                     tracer,
-                    onlinePrewarmingService
+                    onlinePrewarmingService,
+                    crossClusterSearchExtension
                 );
             }