SOLR-17877: Move distributedCollectionCommandRunner to ZkController (#3507)

A refactoring to improve separation-of-concerns.

---------

Co-authored-by: copilot-swe-agent[bot] <[email protected]>
Co-authored-by: David Smiley <[email protected]>

Author: dsmiley

solr/core/src/java/org/apache/solr/cloud/ZkController.java

@@ -66,6 +66,7 @@
 import org.apache.solr.client.solrj.impl.SolrZkClientTimeout;
 import org.apache.solr.client.solrj.impl.ZkClientClusterStateProvider;
 import org.apache.solr.client.solrj.request.CoreAdminRequest.WaitForState;
+import org.apache.solr.cloud.api.collections.DistributedCollectionConfigSetCommandRunner;
 import org.apache.solr.cloud.overseer.ClusterStateMutator;
 import org.apache.solr.cloud.overseer.OverseerAction;
 import org.apache.solr.cloud.overseer.SliceMutator;
@@ -223,6 +224,8 @@ public String toString() {
 
   private final DistributedClusterStateUpdater distributedClusterStateUpdater;
 
+  private final Optional<DistributedCollectionConfigSetCommandRunner> distributedCommandRunner;
+
   private LeaderElector overseerElector;
 
   private Map<String, ReplicateFromLeader> replicateFromLeaders = new ConcurrentHashMap<>();
@@ -370,6 +373,11 @@ public ZkController(
     // Refuse to start if ZK has a non empty /clusterstate.json or a /solr.xml file
     checkNoOldClusterstate(zkClient);
 
+    this.distributedCommandRunner =
+        cloudConfig.getDistributedCollectionConfigSetExecution()
+            ? Optional.of(new DistributedCollectionConfigSetCommandRunner(cc, zkClient))
+            : Optional.empty();
+
     this.overseerRunningMap = Overseer.getRunningMap(zkClient);
     this.overseerCompletedMap = Overseer.getCompletedMap(zkClient);
     this.overseerFailureMap = Overseer.getFailureMap(zkClient);
@@ -681,6 +689,22 @@ public NodesSysPropsCacher getSysPropsCacher() {
     return sysPropsCacher;
   }
 
+  /** Non-empty if the Collection API is executed in a distributed way (Overseer is disabled). */
+  public Optional<DistributedCollectionConfigSetCommandRunner> getDistributedCommandRunner() {
+    return this.distributedCommandRunner;
+  }
+
+  /** Waits for pending tasks to complete. Should be called before {@link #close()}. */
+  public void waitForPendingTasksToComplete() {
+    if (distributedCommandRunner.isPresent()) {
+      // Local (i.e. distributed) Collection API processing
+      distributedCommandRunner.get().stopAndWaitForPendingTasksToComplete();
+    } else {
+      // Overseer based processing
+      getOverseerCollectionQueue().allowOverseerPendingTasksToComplete();
+    }
+  }
+
   private ContextKey closeExistingElectionContext(CoreDescriptor cd, boolean sessionExpired) {
     // look for old context - if we find it, cancel it
     String collection = cd.getCloudDescriptor().getCollectionName();
@@ -779,6 +803,7 @@ public void close() {
     } finally {
 
       sysPropsCacher.close();
+
       customThreadPool.execute(() -> IOUtils.closeQuietly(cloudManager));
       customThreadPool.execute(() -> IOUtils.closeQuietly(cloudSolrClient));
 
@@ -998,7 +1023,9 @@ private void init() {
       checkForExistingEphemeralNode();
       registerLiveNodesListener();
 
-      // start the overseer first as following code may need it's processing
+      // Start the overseer now since the following code may need it's processing.
+      // Note: even when using distributed processing, we still create an Overseer anyway since
+      //    cluster singleton processing is linked to the elected Overseer.
       if (!zkRunOnly) {
         overseerElector = new LeaderElector(zkClient);
         this.overseer =

solr/core/src/java/org/apache/solr/cloud/api/collections/DistributedCollectionCommandContext.java

@@ -27,23 +27,17 @@
 
 public class DistributedCollectionCommandContext implements CollectionCommandContext {
   private final CoreContainer coreContainer;
-  private final DistributedClusterStateUpdater getDistributedClusterStateUpdater;
+  private final DistributedClusterStateUpdater distributedClusterStateUpdater;
   private final ExecutorService executorService;
 
-  private final SolrCloudManager solrCloudManager;
-  private final ZkStateReader zkStateReader;
-
   public DistributedCollectionCommandContext(
       CoreContainer coreContainer, ExecutorService executorService) {
+    // note: coreContainer.getZkController() is not yet instantiated; don't call it right now
     this.coreContainer = coreContainer;
-    this.getDistributedClusterStateUpdater =
+    this.distributedClusterStateUpdater =
         new DistributedClusterStateUpdater(
             coreContainer.getConfig().getCloudConfig().getDistributedClusterStateUpdates());
-    ;
     this.executorService = executorService;
-
-    solrCloudManager = this.coreContainer.getZkController().getSolrCloudManager();
-    zkStateReader = this.coreContainer.getZkController().getZkStateReader();
   }
 
   @Override
@@ -60,7 +54,7 @@ public ShardHandler newShardHandler() {
 
   @Override
   public SolrCloudManager getSolrCloudManager() {
-    return solrCloudManager;
+    return this.coreContainer.getZkController().getSolrCloudManager();
   }
 
   @Override
@@ -70,12 +64,12 @@ public CoreContainer getCoreContainer() {
 
   @Override
   public ZkStateReader getZkStateReader() {
-    return zkStateReader;
+    return this.coreContainer.getZkController().getZkStateReader();
   }
 
   @Override
   public DistributedClusterStateUpdater getDistributedClusterStateUpdater() {
-    return this.getDistributedClusterStateUpdater;
+    return this.distributedClusterStateUpdater;
   }
 
   @Override

solr/core/src/java/org/apache/solr/cloud/api/collections/DistributedCollectionConfigSetCommandRunner.java

@@ -44,6 +44,7 @@
 import org.apache.solr.cloud.ZkDistributedCollectionLockFactory;
 import org.apache.solr.cloud.ZkDistributedConfigSetLockFactory;
 import org.apache.solr.common.SolrException;
+import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.solr.common.cloud.ZkNodeProps;
 import org.apache.solr.common.params.CollectionParams;
 import org.apache.solr.common.params.ConfigSetParams;
@@ -106,7 +107,9 @@ public class DistributedCollectionConfigSetCommandRunner {
 
   private volatile boolean shuttingDown = false;
 
-  public DistributedCollectionConfigSetCommandRunner(CoreContainer coreContainer) {
+  public DistributedCollectionConfigSetCommandRunner(
+      CoreContainer coreContainer, SolrZkClient zkClient) {
+    // note: coreContainer.getZkController() is not yet instantiated; don't call it right now
     this.coreContainer = coreContainer;
 
     if (log.isInfoEnabled()) {
@@ -144,8 +147,7 @@ public DistributedCollectionConfigSetCommandRunner(CoreContainer coreContainer)
         new DistributedCollectionCommandContext(
             this.coreContainer, this.distributedCollectionApiExecutorService);
     commandMapper = new CollApiCmds.CommandMap(ccc);
-    asyncTaskTracker =
-        new DistributedApiAsyncTracker(ccc.getZkStateReader().getZkClient(), ZK_ASYNC_ROOT);
+    asyncTaskTracker = new DistributedApiAsyncTracker(zkClient, ZK_ASYNC_ROOT);
   }
 
   /** See {@link DistributedApiAsyncTracker#getAsyncTaskRequestStatus(String)} */

solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerStatusCmd.java

@@ -165,7 +165,7 @@ public void call(ClusterState state, ZkNodeProps message, NamedList<Object> resu
       throws Exception {
     // If Collection API execution is distributed, we're not running on the Overseer node so can't
     // return any Overseer stats.
-    if (ccc.getCoreContainer().getDistributedCollectionCommandRunner().isPresent()) {
+    if (ccc.getCoreContainer().getZkController().getDistributedCommandRunner().isPresent()) {
       // TODO: introduce a per node status command allowing insight into how Cluster state updates,
       // Collection API and config set API execution went on that node...
       return;

solr/core/src/java/org/apache/solr/core/CoreContainer.java

@@ -45,7 +45,6 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 import java.util.UUID;
@@ -77,9 +76,7 @@
 import org.apache.solr.client.solrj.util.SolrIdentifierValidator;
 import org.apache.solr.cloud.CloudDescriptor;
 import org.apache.solr.cloud.ClusterSingleton;
-import org.apache.solr.cloud.OverseerTaskQueue;
 import org.apache.solr.cloud.ZkController;
-import org.apache.solr.cloud.api.collections.DistributedCollectionConfigSetCommandRunner;
 import org.apache.solr.cluster.events.ClusterEventProducer;
 import org.apache.solr.cluster.events.impl.ClusterEventProducerFactory;
 import org.apache.solr.cluster.placement.PlacementPluginConfig;
@@ -326,14 +323,6 @@ && getZkController().getOverseer() != null
   private ExecutorService coreContainerAsyncTaskExecutor =
       ExecutorUtil.newMDCAwareCachedThreadPool("Core Container Async Task");
 
-  /**
-   * Non-empty if the Collection API is executed in a distributed way and not on Overseer, once the
-   * CoreContainer has been initialized properly, i.e. method {@link #load()} called. Until then it
-   * is null, and it is not expected to be read.
-   */
-  private volatile Optional<DistributedCollectionConfigSetCommandRunner>
-      distributedCollectionCommandRunner;
-
   private enum CoreInitFailedAction {
     fromleader,
     none
@@ -686,7 +675,6 @@ protected CoreContainer(Object testConstructor) {
     cfg = null;
     containerProperties = null;
     replayUpdatesExecutor = null;
-    distributedCollectionCommandRunner = Optional.empty();
     allowPaths = null;
     allowListUrlChecker = null;
     indexSearcherExecutor = null;
@@ -900,19 +888,6 @@ private void loadInternal() {
     createHandler(
         ZK_STATUS_PATH, ZookeeperStatusHandler.class.getName(), ZookeeperStatusHandler.class);
 
-    // CoreContainer is initialized enough at this stage so we can set
-    // distributedCollectionCommandRunner (the construction of
-    // DistributedCollectionConfigSetCommandRunner uses Zookeeper so can't be done from the
-    // CoreContainer constructor because there Zookeeper is not yet ready). Given this is used in
-    // the CollectionsHandler created next line, this is the latest point where
-    // distributedCollectionCommandRunner can be initialized without refactoring this method...
-    // TODO: manage to completely build CoreContainer in the constructor and not in the load()
-    // method... Requires some test refactoring.
-    this.distributedCollectionCommandRunner =
-        isZooKeeperAware() && cfg.getCloudConfig().getDistributedCollectionConfigSetExecution()
-            ? Optional.of(new DistributedCollectionConfigSetCommandRunner(this))
-            : Optional.empty();
-
     collectionsHandler =
         createHandler(
             COLLECTIONS_HANDLER_PATH, cfg.getCollectionsHandlerClass(), CollectionsHandler.class);
@@ -1246,9 +1221,7 @@ protected void configure() {
           throw new SolrException(ErrorCode.SERVER_ERROR, e);
         }
       }
-      if (!distributedCollectionCommandRunner.isPresent()) {
-        zkSys.getZkController().checkOverseerDesignate();
-      }
+      zkSys.getZkController().checkOverseerDesignate();
     }
 
     // This is a bit redundant but these are two distinct concepts for all they're accomplished at
@@ -1318,14 +1291,7 @@ public void shutdown() {
 
     ZkController zkController = getZkController();
     if (zkController != null) {
-      if (distributedCollectionCommandRunner.isPresent()) {
-        // Local (i.e. distributed) Collection API processing
-        distributedCollectionCommandRunner.get().stopAndWaitForPendingTasksToComplete();
-      } else {
-        // Overseer based processing
-        OverseerTaskQueue overseerCollectionQueue = zkController.getOverseerCollectionQueue();
-        overseerCollectionQueue.allowOverseerPendingTasksToComplete();
-      }
+      zkController.waitForPendingTasksToComplete();
     }
     if (log.isInfoEnabled()) {
       log.info("Shutting down CoreContainer instance={}", System.identityHashCode(this));
@@ -2595,11 +2561,6 @@ public PlacementPluginFactory<? extends PlacementPluginConfig> getPlacementPlugi
     return placementPluginFactory;
   }
 
-  public Optional<DistributedCollectionConfigSetCommandRunner>
-      getDistributedCollectionCommandRunner() {
-    return this.distributedCollectionCommandRunner;
-  }
-
   /**
    * A general-purpose HTTP/2 Solr client.
    *