SOLR-7038: Validate config set presence before trying to create a collection

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1654942 13f79535-47bb-0310-9956-ffa450edef68

solr/CHANGES.txt

@@ -553,6 +553,9 @@ Bug Fixes
   shell script, particularly when JAVA_HOME has an invalid location.
   (Shawn Heisey)
 
+* SOLR-7038: Validate the presence of configset before trying to create a collection.
+  (Anshum Gupta, Mark Miller)
+
 Optimizations
 ----------------------
 

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

@@ -2318,6 +2318,13 @@ public class OverseerCollectionProcessor implements Runnable, Closeable {
     if (clusterState.hasCollection(collectionName)) {
       throw new SolrException(ErrorCode.BAD_REQUEST, "collection already exists: " + collectionName);
     }
+
+    String configName = getConfigName(collectionName, message);
+    if (configName == null) {
+      throw new SolrException(ErrorCode.BAD_REQUEST, "No config set found to associate with the collection.");
+    } else if (!validateConfig(configName)) {
+      throw new SolrException(ErrorCode.BAD_REQUEST, "Can not find the specified config set: " + configName);
+    }
     
     try {
       // look at the replication factor and see if it matches reality
@@ -2385,7 +2392,7 @@ public class OverseerCollectionProcessor implements Runnable, Closeable {
       }
       boolean isLegacyCloud =  Overseer.isLegacy(zkStateReader.getClusterProps());
 
-      String configName = createConfNode(collectionName, message, isLegacyCloud);
+      createConfNode(configName, collectionName, isLegacyCloud);
 
       Overseer.getInQueue(zkStateReader.getZkClient()).offer(ZkStateReader.toJSON(message));
 
@@ -2621,26 +2628,38 @@ public class OverseerCollectionProcessor implements Runnable, Closeable {
     } while (srsp != null);
   }
 
-  private String createConfNode(String coll, ZkNodeProps message, boolean isLegacyCloud) throws KeeperException, InterruptedException {
+  private String getConfigName(String coll, ZkNodeProps message) throws KeeperException, InterruptedException {
     String configName = message.getStr(OverseerCollectionProcessor.COLL_CONF);
-    if(configName == null){
+    
+    if (configName == null) {
       // if there is only one conf, use that
-      List<String> configNames=null;
+      List<String> configNames = null;
       try {
         configNames = zkStateReader.getZkClient().getChildren(ZkController.CONFIGS_ZKNODE, null, true);
         if (configNames != null && configNames.size() == 1) {
           configName = configNames.get(0);
           // no config set named, but there is only 1 - use it
           log.info("Only one config set found in zk - using it:" + configName);
-        } else if(configNames.contains(coll)) {
+        } else if (configNames.contains(coll)) {
           configName = coll;
         }
       } catch (KeeperException.NoNodeException e) {
 
       }
-
     }
+    return configName;
+  }
+  
+  private boolean validateConfig(String configName) throws KeeperException, InterruptedException {
+    return zkStateReader.getZkClient().exists(ZkController.CONFIGS_ZKNODE + "/" + configName, true);
+  }
 
+  /**
+   * This doesn't validate the config (path) itself and is just responsible for creating the confNode.
+   * That check should be done before the config node is created.
+   */
+  private void createConfNode(String configName, String coll, boolean isLegacyCloud) throws KeeperException, InterruptedException {
+    
     if (configName != null) {
       String collDir = ZkStateReader.COLLECTIONS_ZKNODE + "/" + coll;
       log.info("creating collections conf node {} ", collDir);
@@ -2657,7 +2676,6 @@ public class OverseerCollectionProcessor implements Runnable, Closeable {
         throw new SolrException(ErrorCode.BAD_REQUEST,"Unable to get config name");
       }
     }
-    return configName;
 
   }
 

solr/core/src/test/org/apache/solr/cloud/OverseerCollectionProcessorTest.java

@@ -338,6 +338,8 @@ public class OverseerCollectionProcessorTest extends SolrTestCaseJ4 {
       }
     }).anyTimes();
     
+    zkMap.put("/configs/myconfig", null);
+    
     return liveNodes;
   }