SOLR-15055: Re-implement 'withCollection'.

Author: sigram

solr/CHANGES.txt

@@ -25,6 +25,9 @@ New Features
 
 * SOLR-15019: Replica placement API needs a way to fetch existing replica metrics. (ab, ilan)
 
+* SOLR-15055: Re-implement 'withCollection'. This also adds the placement plugin support
+  for rejecting replica / collection deletions that would violate placement constraints. (ab, ilan)
+
 Improvements
 ----------------------
 * LUCENE-8984: MoreLikeThis MLT is biased for uncommon fields (Andy Hind via Anshum Gupta)

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

@@ -29,7 +29,6 @@
 import java.util.Set;
 
 import org.apache.commons.lang3.StringUtils;
-import org.apache.solr.cloud.api.collections.OverseerCollectionMessageHandler;
 import org.apache.solr.cloud.overseer.ClusterStateMutator;
 import org.apache.solr.cloud.overseer.CollectionMutator;
 import org.apache.solr.cloud.overseer.SliceMutator;
@@ -40,6 +39,7 @@
 import org.apache.solr.common.cloud.Slice;
 import org.apache.solr.common.cloud.ZkNodeProps;
 import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.CollectionAdminParams;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -74,8 +74,8 @@ class ExclusiveSliceProperty {
   ExclusiveSliceProperty(ClusterState clusterState, ZkNodeProps message) {
     this.clusterState = clusterState;
     String tmp = message.getStr(ZkStateReader.PROPERTY_PROP);
-    if (StringUtils.startsWith(tmp, OverseerCollectionMessageHandler.COLL_PROP_PREFIX) == false) {
-      tmp = OverseerCollectionMessageHandler.COLL_PROP_PREFIX + tmp;
+    if (!StringUtils.startsWith(tmp, CollectionAdminParams.PROPERTY_PREFIX)) {
+      tmp = CollectionAdminParams.PROPERTY_PREFIX + tmp;
     }
     this.property = tmp.toLowerCase(Locale.ROOT);
     collectionName = message.getStr(ZkStateReader.COLLECTION_PROP);

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

@@ -45,7 +45,6 @@
 import org.apache.solr.cloud.ActiveReplicaWatcher;
 import org.apache.solr.cloud.Overseer;
 import org.apache.solr.cloud.api.collections.OverseerCollectionMessageHandler.ShardRequestTracker;
-import org.apache.solr.cluster.placement.PlacementPlugin;
 import org.apache.solr.common.SolrCloseableLatch;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.cloud.ClusterState;
@@ -61,6 +60,7 @@
 import org.apache.solr.common.params.ShardParams;
 import org.apache.solr.common.util.NamedList;
 import org.apache.solr.common.util.Utils;
+import org.apache.solr.core.CoreContainer;
 import org.apache.solr.handler.component.ShardHandler;
 import org.apache.zookeeper.KeeperException;
 import org.slf4j.Logger;
@@ -141,7 +141,7 @@ List<ZkNodeProps> addReplica(ClusterState clusterState, ZkNodeProps message, @Su
     }
 
     List<CreateReplica> createReplicas = buildReplicaPositions(ocmh.cloudManager, clusterState, collectionName, message, replicaTypesVsCount,
-        ocmh.overseer.getCoreContainer().getPlacementPluginFactory().createPluginInstance())
+        ocmh.overseer.getCoreContainer())
           .stream()
           .map(replicaPosition -> assignReplicaDetails(ocmh.cloudManager, clusterState, message, replicaPosition))
           .collect(Collectors.toList());
@@ -302,7 +302,7 @@ public static CreateReplica assignReplicaDetails(SolrCloudManager cloudManager,
   public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
                                                             String collectionName, ZkNodeProps message,
                                                             EnumMap<Replica.Type, Integer> replicaTypeVsCount,
-                                                            PlacementPlugin placementPlugin) throws IOException, InterruptedException {
+                                                            CoreContainer coreContainer) throws IOException, InterruptedException {
     boolean skipCreateReplicaInClusterState = message.getBool(SKIP_CREATE_REPLICA_IN_CLUSTER_STATE, false);
     boolean skipNodeAssignment = message.getBool(CollectionAdminParams.SKIP_NODE_ASSIGNMENT, false);
     String sliceName = message.getStr(SHARD_ID_PROP);
@@ -326,7 +326,7 @@ public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloud
     if (!skipCreateReplicaInClusterState && !skipNodeAssignment) {
 
       positions = Assign.getNodesForNewReplicas(clusterState, collection.getName(), sliceName, numNrtReplicas,
-                    numTlogReplicas, numPullReplicas, createNodeSetStr, cloudManager, placementPlugin);
+                    numTlogReplicas, numPullReplicas, createNodeSetStr, cloudManager, coreContainer);
     }
 
     if (positions == null)  {

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

@@ -51,6 +51,7 @@
 import org.apache.solr.common.cloud.ZkNodeProps;
 import org.apache.solr.common.cloud.ZkStateReader;
 import org.apache.solr.common.util.StrUtils;
+import org.apache.solr.core.CoreContainer;
 import org.apache.solr.util.NumberUtils;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
@@ -270,7 +271,7 @@ public int weight() {
   public static List<ReplicaPosition> getNodesForNewReplicas(ClusterState clusterState, String collectionName,
                                                           String shard, int nrtReplicas, int tlogReplicas, int pullReplicas,
                                                           Object createNodeSet, SolrCloudManager cloudManager,
-                                                          PlacementPlugin placementPlugin) throws IOException, InterruptedException, AssignmentException {
+                                                          CoreContainer coreContainer) throws IOException, InterruptedException, AssignmentException {
     log.debug("getNodesForNewReplicas() shard: {} , nrtReplicas : {} , tlogReplicas: {} , pullReplicas: {} , createNodeSet {}"
         , shard, nrtReplicas, tlogReplicas, pullReplicas, createNodeSet);
     DocCollection coll = clusterState.getCollection(collectionName);
@@ -296,7 +297,7 @@ public static List<ReplicaPosition> getNodesForNewReplicas(ClusterState clusterS
         .assignPullReplicas(pullReplicas)
         .onNodes(createNodeList)
         .build();
-    AssignStrategy assignStrategy = createAssignStrategy(placementPlugin, clusterState, coll);
+    AssignStrategy assignStrategy = createAssignStrategy(coreContainer, clusterState, coll);
     return assignStrategy.assign(cloudManager, assignRequest);
   }
 
@@ -379,9 +380,46 @@ public AssignmentException(String message, Throwable cause, boolean enableSuppre
     }
   }
 
+  /**
+   * Strategy for assigning replicas to nodes.
+   */
   public interface AssignStrategy {
+
+    /**
+     * Assign new replicas to nodes.
+     * @param solrCloudManager current instance of {@link SolrCloudManager}.
+     * @param assignRequest assign request.
+     * @return list of {@link ReplicaPosition}-s for new replicas.
+     * @throws AssignmentException when assignment request cannot produce any valid assignments.
+     */
     List<ReplicaPosition> assign(SolrCloudManager solrCloudManager, AssignRequest assignRequest)
-        throws Assign.AssignmentException, IOException, InterruptedException;
+        throws AssignmentException, IOException, InterruptedException;
+
+    /**
+     * Verify that deleting a collection doesn't violate the replica assignment constraints.
+     * @param solrCloudManager current instance of {@link SolrCloudManager}.
+     * @param collection collection to delete.
+     * @throws AssignmentException when deleting the collection would violate replica assignment constraints.
+     * @throws IOException on general errors.
+     */
+    default void verifyDeleteCollection(SolrCloudManager solrCloudManager, DocCollection collection)
+        throws AssignmentException, IOException, InterruptedException {
+
+    }
+
+    /**
+     * Verify that deleting these replicas doesn't violate the replica assignment constraints.
+     * @param solrCloudManager current instance of {@link SolrCloudManager}.
+     * @param collection collection to delete replicas from.
+     * @param shardName shard name.
+     * @param replicas replicas to delete.
+     * @throws AssignmentException when deleting the replicas would violate replica assignment constraints.
+     * @throws IOException on general errors.
+     */
+    default void verifyDeleteReplicas(SolrCloudManager solrCloudManager, DocCollection collection, String shardName, Set<Replica> replicas)
+        throws AssignmentException, IOException, InterruptedException {
+
+    }
   }
 
   public static class AssignRequest {
@@ -495,7 +533,8 @@ private ImmutableMap<Replica.Type, Integer> countsPerReplicaType(AssignRequest a
    * <p>If {@link PlacementPlugin} instance is null this call will return {@link LegacyAssignStrategy}, otherwise
    * {@link PlacementPluginAssignStrategy} will be used.</p>
    */
-  public static AssignStrategy createAssignStrategy(PlacementPlugin placementPlugin, ClusterState clusterState, DocCollection collection) {
+  public static AssignStrategy createAssignStrategy(CoreContainer coreContainer, ClusterState clusterState, DocCollection collection) {
+    PlacementPlugin placementPlugin = coreContainer.getPlacementPluginFactory().createPluginInstance();
     if (placementPlugin != null) {
       // If a cluster wide placement plugin is configured (and that's the only way to define a placement plugin)
       return new PlacementPluginAssignStrategy(collection, placementPlugin);

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

@@ -41,7 +41,6 @@
 import org.apache.solr.cloud.ZkController;
 import org.apache.solr.cloud.api.collections.OverseerCollectionMessageHandler.ShardRequestTracker;
 import org.apache.solr.cloud.overseer.ClusterStateMutator;
-import org.apache.solr.cluster.placement.PlacementPlugin;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.SolrException.ErrorCode;
 import org.apache.solr.common.cloud.Aliases;
@@ -63,6 +62,7 @@
 import org.apache.solr.common.util.SimpleOrderedMap;
 import org.apache.solr.common.util.TimeSource;
 import org.apache.solr.common.util.Utils;
+import org.apache.solr.core.CoreContainer;
 import org.apache.solr.handler.component.ShardHandler;
 import org.apache.solr.handler.component.ShardRequest;
 import org.apache.solr.util.TimeOut;
@@ -169,8 +169,8 @@ public void call(ClusterState clusterState, ZkNodeProps message, @SuppressWarnin
 
       List<ReplicaPosition> replicaPositions = null;
       try {
-        replicaPositions = buildReplicaPositions(ocmh.cloudManager, clusterState, clusterState.getCollection(collectionName),
-            message, shardNames, ocmh.overseer.getCoreContainer().getPlacementPluginFactory().createPluginInstance());
+        replicaPositions = buildReplicaPositions(ocmh.overseer.getCoreContainer(), ocmh.cloudManager, clusterState, clusterState.getCollection(collectionName),
+            message, shardNames);
       } catch (Assign.AssignmentException e) {
         ZkNodeProps deleteMessage = new ZkNodeProps("name", collectionName);
         new DeleteCollectionCmd(ocmh).call(clusterState, deleteMessage, results);
@@ -288,10 +288,10 @@ public void call(ClusterState clusterState, ZkNodeProps message, @SuppressWarnin
     }
   }
 
-  private static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
+  private static List<ReplicaPosition> buildReplicaPositions(CoreContainer coreContainer, SolrCloudManager cloudManager, ClusterState clusterState,
                                                              DocCollection docCollection,
                                                              ZkNodeProps message,
-                                                             List<String> shardNames, PlacementPlugin placementPlugin) throws IOException, InterruptedException, Assign.AssignmentException {
+                                                             List<String> shardNames) throws IOException, InterruptedException, Assign.AssignmentException {
     final String collectionName = message.getStr(NAME);
     // look at the replication factor and see if it matches reality
     // if it does not, find best nodes to create more cores
@@ -330,7 +330,7 @@ private static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager clou
           .assignPullReplicas(numPullReplicas)
           .onNodes(nodeList)
           .build();
-      Assign.AssignStrategy assignStrategy = Assign.createAssignStrategy(placementPlugin, clusterState, docCollection);
+      Assign.AssignStrategy assignStrategy = Assign.createAssignStrategy(coreContainer, clusterState, docCollection);
       replicaPositions = assignStrategy.assign(cloudManager, assignRequest);
     }
     return replicaPositions;

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

@@ -92,6 +92,13 @@ public void call(ClusterState state, ZkNodeProps message, @SuppressWarnings({"ra
       collection = extCollection;
     }
 
+    // verify the placement modifications caused by the deletion are allowed
+    DocCollection coll = state.getCollectionOrNull(collection);
+    if (coll != null) {
+      Assign.AssignStrategy assignStrategy = Assign.createAssignStrategy(ocmh.overseer.getCoreContainer(), state, coll);
+      assignStrategy.verifyDeleteCollection(ocmh.cloudManager, coll);
+    }
+
     final boolean deleteHistory = message.getBool(CoreAdminParams.DELETE_METRICS_HISTORY, true);
 
     boolean removeCounterNode = true;

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

@@ -18,6 +18,7 @@
 package org.apache.solr.cloud.api.collections;
 
 
+import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
 import java.util.List;
@@ -98,7 +99,7 @@ static void cleanupReplicas(@SuppressWarnings({"rawtypes"})NamedList results,
                               List<ZkNodeProps> sourceReplicas,
                               OverseerCollectionMessageHandler ocmh,
                               String node,
-                              String async) throws InterruptedException {
+                              String async) throws IOException, InterruptedException {
     CountDownLatch cleanupLatch = new CountDownLatch(sourceReplicas.size());
     for (ZkNodeProps sourceReplica : sourceReplicas) {
       String coll = sourceReplica.getStr(COLLECTION_PROP);