You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by da...@apache.org on 2018/01/23 10:30:53 UTC
[23/41] lucene-solr:jira/solr-11702: SOLR-11817: Move Collections API
classes to it's own package
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
deleted file mode 100644
index 2171c60..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
+++ /dev/null
@@ -1,533 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.solr.cloud;
-
-
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
-import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
-import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
-import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
-import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
-import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
-import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
-import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
-import org.apache.solr.cloud.overseer.ClusterStateMutator;
-import org.apache.solr.common.SolrException;
-import org.apache.solr.common.SolrException.ErrorCode;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocRouter;
-import org.apache.solr.common.cloud.ImplicitDocRouter;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.ReplicaPosition;
-import org.apache.solr.common.cloud.ZkConfigManager;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.cloud.ZooKeeperException;
-import org.apache.solr.common.params.CollectionAdminParams;
-import org.apache.solr.common.params.CommonAdminParams;
-import org.apache.solr.common.params.CoreAdminParams;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.NamedList;
-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.handler.admin.ConfigSetsHandlerApi;
-import org.apache.solr.handler.component.ShardHandler;
-import org.apache.solr.handler.component.ShardRequest;
-import org.apache.solr.util.TimeOut;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.KeeperException.NoNodeException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_CONF;
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.CREATE_NODE_SET;
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.NUM_SLICES;
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.RANDOM;
-import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
-import static org.apache.solr.common.cloud.ZkStateReader.NRT_REPLICAS;
-import static org.apache.solr.common.cloud.ZkStateReader.PULL_REPLICAS;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
-import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-import static org.apache.solr.common.params.CommonAdminParams.WAIT_FOR_FINAL_STATE;
-import static org.apache.solr.common.params.CommonParams.NAME;
-import static org.apache.solr.common.util.StrUtils.formatString;
-
-public class CreateCollectionCmd implements Cmd {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final OverseerCollectionMessageHandler ocmh;
- private final TimeSource timeSource;
- private final DistribStateManager stateManager;
-
- public CreateCollectionCmd(OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- this.stateManager = ocmh.cloudManager.getDistribStateManager();
- this.timeSource = ocmh.cloudManager.getTimeSource();
- }
-
- @Override
- public void call(ClusterState clusterState, ZkNodeProps message, NamedList results) throws Exception {
- final String collectionName = message.getStr(NAME);
- final boolean waitForFinalState = message.getBool(WAIT_FOR_FINAL_STATE, false);
- log.info("Create collection {}", collectionName);
- if (clusterState.hasCollection(collectionName)) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "collection already exists: " + collectionName);
- }
-
- String configName = getConfigName(collectionName, message);
- if (configName == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No config set found to associate with the collection.");
- }
-
- ocmh.validateConfigOrThrowSolrException(configName);
- AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
-
- try {
-
- final String async = message.getStr(ASYNC);
-
- List<String> nodeList = new ArrayList<>();
- List<String> shardNames = new ArrayList<>();
- List<ReplicaPosition> replicaPositions = buildReplicaPositions(ocmh.cloudManager, clusterState, message,
- nodeList, shardNames, sessionWrapper);
- ZkStateReader zkStateReader = ocmh.zkStateReader;
- boolean isLegacyCloud = Overseer.isLegacy(zkStateReader);
-
- ocmh.createConfNode(stateManager, configName, collectionName, isLegacyCloud);
-
- Map<String,String> collectionParams = new HashMap<>();
- Map<String,Object> collectionProps = message.getProperties();
- for (String propName : collectionProps.keySet()) {
- if (propName.startsWith(ZkController.COLLECTION_PARAM_PREFIX)) {
- collectionParams.put(propName.substring(ZkController.COLLECTION_PARAM_PREFIX.length()), (String) collectionProps.get(propName));
- }
- }
-
- createCollectionZkNode(stateManager, collectionName, collectionParams);
-
- Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(message));
-
- // wait for a while until we don't see the collection
- TimeOut waitUntil = new TimeOut(30, TimeUnit.SECONDS, timeSource);
- boolean created = false;
- while (! waitUntil.hasTimedOut()) {
- waitUntil.sleep(100);
- created = ocmh.cloudManager.getClusterStateProvider().getClusterState().hasCollection(collectionName);
- if(created) break;
- }
- if (!created)
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Could not fully create collection: " + collectionName);
-
- if (nodeList.isEmpty()) {
- log.debug("Finished create command for collection: {}", collectionName);
- return;
- }
-
- // For tracking async calls.
- Map<String, String> requestMap = new HashMap<>();
-
-
- log.debug(formatString("Creating SolrCores for new collection {0}, shardNames {1} , message : {2}",
- collectionName, shardNames, message));
- Map<String,ShardRequest> coresToCreate = new LinkedHashMap<>();
- ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
- for (ReplicaPosition replicaPosition : replicaPositions) {
- String nodeName = replicaPosition.node;
- String coreName = Assign.buildSolrCoreName(ocmh.cloudManager.getDistribStateManager(),
- ocmh.cloudManager.getClusterStateProvider().getClusterState().getCollection(collectionName),
- replicaPosition.shard, replicaPosition.type, true);
- log.debug(formatString("Creating core {0} as part of shard {1} of collection {2} on {3}"
- , coreName, replicaPosition.shard, collectionName, nodeName));
-
-
- String baseUrl = zkStateReader.getBaseUrlForNodeName(nodeName);
- //in the new mode, create the replica in clusterstate prior to creating the core.
- // Otherwise the core creation fails
- if (!isLegacyCloud) {
- ZkNodeProps props = new ZkNodeProps(
- Overseer.QUEUE_OPERATION, ADDREPLICA.toString(),
- ZkStateReader.COLLECTION_PROP, collectionName,
- ZkStateReader.SHARD_ID_PROP, replicaPosition.shard,
- ZkStateReader.CORE_NAME_PROP, coreName,
- ZkStateReader.STATE_PROP, Replica.State.DOWN.toString(),
- ZkStateReader.BASE_URL_PROP, baseUrl,
- ZkStateReader.REPLICA_TYPE, replicaPosition.type.name(),
- CommonAdminParams.WAIT_FOR_FINAL_STATE, Boolean.toString(waitForFinalState));
- Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
- }
-
- // Need to create new params for each request
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.CREATE.toString());
-
- params.set(CoreAdminParams.NAME, coreName);
- params.set(COLL_CONF, configName);
- params.set(CoreAdminParams.COLLECTION, collectionName);
- params.set(CoreAdminParams.SHARD, replicaPosition.shard);
- params.set(ZkStateReader.NUM_SHARDS_PROP, shardNames.size());
- params.set(CoreAdminParams.NEW_COLLECTION, "true");
- params.set(CoreAdminParams.REPLICA_TYPE, replicaPosition.type.name());
-
- if (async != null) {
- String coreAdminAsyncId = async + Math.abs(System.nanoTime());
- params.add(ASYNC, coreAdminAsyncId);
- requestMap.put(nodeName, coreAdminAsyncId);
- }
- ocmh.addPropertyParams(message, params);
-
- ShardRequest sreq = new ShardRequest();
- sreq.nodeName = nodeName;
- params.set("qt", ocmh.adminPath);
- sreq.purpose = 1;
- sreq.shards = new String[]{baseUrl};
- sreq.actualShards = sreq.shards;
- sreq.params = params;
-
- if (isLegacyCloud) {
- shardHandler.submit(sreq, sreq.shards[0], sreq.params);
- } else {
- coresToCreate.put(coreName, sreq);
- }
- }
-
- if(!isLegacyCloud) {
- // wait for all replica entries to be created
- Map<String, Replica> replicas = ocmh.waitToSeeReplicasInState(collectionName, coresToCreate.keySet());
- for (Map.Entry<String, ShardRequest> e : coresToCreate.entrySet()) {
- ShardRequest sreq = e.getValue();
- sreq.params.set(CoreAdminParams.CORE_NODE_NAME, replicas.get(e.getKey()).getName());
- shardHandler.submit(sreq, sreq.shards[0], sreq.params);
- }
- }
-
- ocmh.processResponses(results, shardHandler, false, null, async, requestMap, Collections.emptySet());
- if(results.get("failure") != null && ((SimpleOrderedMap)results.get("failure")).size() > 0) {
- // Let's cleanup as we hit an exception
- // We shouldn't be passing 'results' here for the cleanup as the response would then contain 'success'
- // element, which may be interpreted by the user as a positive ack
- ocmh.cleanupCollection(collectionName, new NamedList());
- log.info("Cleaned up artifacts for failed create collection for [{}]", collectionName);
- } else {
- log.debug("Finished create command on all shards for collection: {}", collectionName);
-
- // Emit a warning about production use of data driven functionality
- boolean defaultConfigSetUsed = message.getStr(COLL_CONF) == null ||
- message.getStr(COLL_CONF).equals(ConfigSetsHandlerApi.DEFAULT_CONFIGSET_NAME);
- if (defaultConfigSetUsed) {
- results.add("warning", "Using _default configset. Data driven schema functionality"
- + " is enabled by default, which is NOT RECOMMENDED for production use. To turn it off:"
- + " curl http://{host:port}/solr/" + collectionName + "/config -d '{\"set-user-property\": {\"update.autoCreateFields\":\"false\"}}'");
- }
- }
- } catch (SolrException ex) {
- throw ex;
- } catch (Exception ex) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, null, ex);
- } finally {
- if (sessionWrapper.get() != null) sessionWrapper.get().release();
- }
- }
-
- public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
- ZkNodeProps message,
- List<String> nodeList, List<String> shardNames,
- AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper) throws IOException, InterruptedException {
- 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
- int numTlogReplicas = message.getInt(TLOG_REPLICAS, 0);
- int numNrtReplicas = message.getInt(NRT_REPLICAS, message.getInt(REPLICATION_FACTOR, numTlogReplicas>0?0:1));
- int numPullReplicas = message.getInt(PULL_REPLICAS, 0);
- AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
- String policy = message.getStr(Policy.POLICY);
- boolean usePolicyFramework = !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty() || policy != null;
-
- Integer numSlices = message.getInt(NUM_SLICES, null);
- String router = message.getStr("router.name", DocRouter.DEFAULT_NAME);
- if(ImplicitDocRouter.NAME.equals(router)){
- ClusterStateMutator.getShardNames(shardNames, message.getStr("shards", null));
- numSlices = shardNames.size();
- } else {
- if (numSlices == null ) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NUM_SLICES + " is a required param (when using CompositeId router).");
- }
- if (numSlices <= 0) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NUM_SLICES + " must be > 0");
- }
- ClusterStateMutator.getShardNames(numSlices, shardNames);
- }
-
- int maxShardsPerNode = message.getInt(MAX_SHARDS_PER_NODE, 1);
- if (usePolicyFramework && message.getStr(MAX_SHARDS_PER_NODE) != null && maxShardsPerNode > 0) {
- throw new SolrException(ErrorCode.BAD_REQUEST, "'maxShardsPerNode>0' is not supported when autoScaling policies are used");
- }
- if (maxShardsPerNode == -1 || usePolicyFramework) maxShardsPerNode = Integer.MAX_VALUE;
- if (numNrtReplicas + numTlogReplicas <= 0) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NRT_REPLICAS + " + " + TLOG_REPLICAS + " must be greater than 0");
- }
-
- // we need to look at every node and see how many cores it serves
- // add our new cores to existing nodes serving the least number of cores
- // but (for now) require that each core goes on a distinct node.
-
- List<ReplicaPosition> replicaPositions;
- nodeList.addAll(Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM));
- if (nodeList.isEmpty()) {
- log.warn("It is unusual to create a collection ("+collectionName+") without cores.");
-
- replicaPositions = new ArrayList<>();
- } else {
- int totalNumReplicas = numNrtReplicas + numTlogReplicas + numPullReplicas;
- if (totalNumReplicas > nodeList.size()) {
- log.warn("Specified number of replicas of "
- + totalNumReplicas
- + " on collection "
- + collectionName
- + " is higher than the number of Solr instances currently live or live and part of your " + CREATE_NODE_SET + "("
- + nodeList.size()
- + "). It's unusual to run two replica of the same slice on the same Solr-instance.");
- }
-
- int maxShardsAllowedToCreate = maxShardsPerNode == Integer.MAX_VALUE ?
- Integer.MAX_VALUE :
- maxShardsPerNode * nodeList.size();
- int requestedShardsToCreate = numSlices * totalNumReplicas;
- if (maxShardsAllowedToCreate < requestedShardsToCreate) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Cannot create collection " + collectionName + ". Value of "
- + MAX_SHARDS_PER_NODE + " is " + maxShardsPerNode
- + ", and the number of nodes currently live or live and part of your "+CREATE_NODE_SET+" is " + nodeList.size()
- + ". This allows a maximum of " + maxShardsAllowedToCreate
- + " to be created. Value of " + NUM_SLICES + " is " + numSlices
- + ", value of " + NRT_REPLICAS + " is " + numNrtReplicas
- + ", value of " + TLOG_REPLICAS + " is " + numTlogReplicas
- + " and value of " + PULL_REPLICAS + " is " + numPullReplicas
- + ". This requires " + requestedShardsToCreate
- + " shards to be created (higher than the allowed number)");
- }
- replicaPositions = Assign.identifyNodes(cloudManager
- , clusterState, nodeList, collectionName, message, shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas);
- sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
- }
- return replicaPositions;
- }
-
- String getConfigName(String coll, ZkNodeProps message) throws KeeperException, InterruptedException {
- String configName = message.getStr(COLL_CONF);
-
- if (configName == null) {
- // if there is only one conf, use that
- List<String> configNames = null;
- try {
- configNames = ocmh.zkStateReader.getZkClient().getChildren(ZkConfigManager.CONFIGS_ZKNODE, null, true);
- if (configNames.contains(ConfigSetsHandlerApi.DEFAULT_CONFIGSET_NAME)) {
- if (!CollectionAdminParams.SYSTEM_COLL.equals(coll)) {
- copyDefaultConfigSetTo(configNames, coll);
- }
- return coll;
- } else 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);
- }
- } catch (KeeperException.NoNodeException e) {
-
- }
- }
- return "".equals(configName)? null: configName;
- }
-
- /**
- * Copies the _default configset to the specified configset name (overwrites if pre-existing)
- */
- private void copyDefaultConfigSetTo(List<String> configNames, String targetConfig) {
- ZkConfigManager configManager = new ZkConfigManager(ocmh.zkStateReader.getZkClient());
-
- // if a configset named coll exists, delete the configset so that _default can be copied over
- if (configNames.contains(targetConfig)) {
- log.info("There exists a configset by the same name as the collection we're trying to create: " + targetConfig +
- ", deleting it so that we can copy the _default configs over and create the collection.");
- try {
- configManager.deleteConfigDir(targetConfig);
- } catch (Exception e) {
- throw new SolrException(ErrorCode.INVALID_STATE, "Error while deleting configset: " + targetConfig, e);
- }
- } else {
- log.info("Only _default config set found, using it.");
- }
- // Copy _default into targetConfig
- try {
- configManager.copyConfigDir(ConfigSetsHandlerApi.DEFAULT_CONFIGSET_NAME, targetConfig, new HashSet<>());
- } catch (Exception e) {
- throw new SolrException(ErrorCode.INVALID_STATE, "Error while copying _default to " + targetConfig, e);
- }
- }
-
- public static void createCollectionZkNode(DistribStateManager stateManager, String collection, Map<String,String> params) {
- log.debug("Check for collection zkNode:" + collection);
- String collectionPath = ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection;
-
- try {
- if (!stateManager.hasData(collectionPath)) {
- log.debug("Creating collection in ZooKeeper:" + collection);
-
- try {
- Map<String,Object> collectionProps = new HashMap<>();
-
- // TODO: if collection.configName isn't set, and there isn't already a conf in zk, just use that?
- String defaultConfigName = System.getProperty(ZkController.COLLECTION_PARAM_PREFIX + ZkController.CONFIGNAME_PROP, collection);
-
- if (params.size() > 0) {
- collectionProps.putAll(params);
- // if the config name wasn't passed in, use the default
- if (!collectionProps.containsKey(ZkController.CONFIGNAME_PROP)) {
- // users can create the collection node and conf link ahead of time, or this may return another option
- getConfName(stateManager, collection, collectionPath, collectionProps);
- }
-
- } else if (System.getProperty("bootstrap_confdir") != null) {
- // if we are bootstrapping a collection, default the config for
- // a new collection to the collection we are bootstrapping
- log.info("Setting config for collection:" + collection + " to " + defaultConfigName);
-
- Properties sysProps = System.getProperties();
- for (String sprop : System.getProperties().stringPropertyNames()) {
- if (sprop.startsWith(ZkController.COLLECTION_PARAM_PREFIX)) {
- collectionProps.put(sprop.substring(ZkController.COLLECTION_PARAM_PREFIX.length()), sysProps.getProperty(sprop));
- }
- }
-
- // if the config name wasn't passed in, use the default
- if (!collectionProps.containsKey(ZkController.CONFIGNAME_PROP))
- collectionProps.put(ZkController.CONFIGNAME_PROP, defaultConfigName);
-
- } else if (Boolean.getBoolean("bootstrap_conf")) {
- // the conf name should should be the collection name of this core
- collectionProps.put(ZkController.CONFIGNAME_PROP, collection);
- } else {
- getConfName(stateManager, collection, collectionPath, collectionProps);
- }
-
- collectionProps.remove(ZkStateReader.NUM_SHARDS_PROP); // we don't put numShards in the collections properties
-
- ZkNodeProps zkProps = new ZkNodeProps(collectionProps);
- stateManager.makePath(collectionPath, Utils.toJSON(zkProps), CreateMode.PERSISTENT, false);
-
- } catch (KeeperException e) {
- // it's okay if the node already exists
- if (e.code() != KeeperException.Code.NODEEXISTS) {
- throw e;
- }
- } catch (AlreadyExistsException e) {
- // it's okay if the node already exists
- }
- } else {
- log.debug("Collection zkNode exists");
- }
-
- } catch (KeeperException e) {
- // it's okay if another beats us creating the node
- if (e.code() == KeeperException.Code.NODEEXISTS) {
- return;
- }
- throw new SolrException(ErrorCode.SERVER_ERROR, "Error creating collection node in Zookeeper", e);
- } catch (IOException e) {
- throw new SolrException(ErrorCode.SERVER_ERROR, "Error creating collection node in Zookeeper", e);
- } catch (InterruptedException e) {
- Thread.interrupted();
- throw new SolrException(ErrorCode.SERVER_ERROR, "Error creating collection node in Zookeeper", e);
- }
-
- }
-
- private static void getConfName(DistribStateManager stateManager, String collection, String collectionPath, Map<String,Object> collectionProps) throws IOException,
- KeeperException, InterruptedException {
- // check for configName
- log.debug("Looking for collection configName");
- if (collectionProps.containsKey("configName")) {
- log.info("configName was passed as a param {}", collectionProps.get("configName"));
- return;
- }
-
- List<String> configNames = null;
- int retry = 1;
- int retryLimt = 6;
- for (; retry < retryLimt; retry++) {
- if (stateManager.hasData(collectionPath)) {
- VersionedData data = stateManager.getData(collectionPath);
- ZkNodeProps cProps = ZkNodeProps.load(data.getData());
- if (cProps.containsKey(ZkController.CONFIGNAME_PROP)) {
- break;
- }
- }
-
- try {
- configNames = stateManager.listData(ZkConfigManager.CONFIGS_ZKNODE);
- } catch (NoSuchElementException | NoNodeException e) {
- // just keep trying
- }
-
- // check if there's a config set with the same name as the collection
- if (configNames != null && configNames.contains(collection)) {
- log.info(
- "Could not find explicit collection configName, but found config name matching collection name - using that set.");
- collectionProps.put(ZkController.CONFIGNAME_PROP, collection);
- break;
- }
- // if _default exists, use that
- if (configNames != null && configNames.contains(ConfigSetsHandlerApi.DEFAULT_CONFIGSET_NAME)) {
- log.info(
- "Could not find explicit collection configName, but found _default config set - using that set.");
- collectionProps.put(ZkController.CONFIGNAME_PROP, ConfigSetsHandlerApi.DEFAULT_CONFIGSET_NAME);
- break;
- }
- // if there is only one conf, use that
- if (configNames != null && configNames.size() == 1) {
- // no config set named, but there is only 1 - use it
- log.info("Only one config set found in zk - using it:" + configNames.get(0));
- collectionProps.put(ZkController.CONFIGNAME_PROP, configNames.get(0));
- break;
- }
-
- log.info("Could not find collection configName - pausing for 3 seconds and trying again - try: " + retry);
- Thread.sleep(3000);
- }
- if (retry == retryLimt) {
- log.error("Could not find configName for collection " + collection);
- throw new ZooKeeperException(
- SolrException.ErrorCode.SERVER_ERROR,
- "Could not find configName for collection " + collection + " found:" + configNames);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
deleted file mode 100644
index c6afdcc..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
+++ /dev/null
@@ -1,191 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.solr.cloud;
-
-
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-import com.google.common.collect.ImmutableMap;
-import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
-import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
-import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
-import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
-import org.apache.solr.common.SolrCloseableLatch;
-import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.ReplicaPosition;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.CommonAdminParams;
-import org.apache.solr.common.params.CoreAdminParams;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.SimpleOrderedMap;
-import org.apache.solr.common.util.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.cloud.Assign.getNodesForNewReplicas;
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.RANDOM;
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.NRT_REPLICAS;
-import static org.apache.solr.common.cloud.ZkStateReader.PULL_REPLICAS;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-
-public class CreateShardCmd implements Cmd {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final OverseerCollectionMessageHandler ocmh;
-
- public CreateShardCmd(OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- }
-
- @Override
- public void call(ClusterState clusterState, ZkNodeProps message, NamedList results) throws Exception {
- String collectionName = message.getStr(COLLECTION_PROP);
- String sliceName = message.getStr(SHARD_ID_PROP);
- boolean waitForFinalState = message.getBool(CommonAdminParams.WAIT_FOR_FINAL_STATE, false);
-
- log.info("Create shard invoked: {}", message);
- if (collectionName == null || sliceName == null)
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "'collection' and 'shard' are required parameters");
-
- DocCollection collection = clusterState.getCollection(collectionName);
-
- ZkStateReader zkStateReader = ocmh.zkStateReader;
- AtomicReference<PolicyHelper.SessionWrapper> sessionWrapper = new AtomicReference<>();
- SolrCloseableLatch countDownLatch;
- try {
- List<ReplicaPosition> positions = buildReplicaPositions(ocmh.cloudManager, clusterState, collectionName, message, sessionWrapper);
- Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(message));
- // wait for a while until we see the shard
- ocmh.waitForNewShard(collectionName, sliceName);
-
- String async = message.getStr(ASYNC);
- countDownLatch = new SolrCloseableLatch(positions.size(), ocmh);
- for (ReplicaPosition position : positions) {
- String nodeName = position.node;
- String coreName = Assign.buildSolrCoreName(ocmh.cloudManager.getDistribStateManager(), collection, sliceName, position.type);
- log.info("Creating replica " + coreName + " as part of slice " + sliceName + " of collection " + collectionName
- + " on " + nodeName);
-
- // Need to create new params for each request
- ZkNodeProps addReplicasProps = new ZkNodeProps(
- COLLECTION_PROP, collectionName,
- SHARD_ID_PROP, sliceName,
- ZkStateReader.REPLICA_TYPE, position.type.name(),
- CoreAdminParams.NODE, nodeName,
- CoreAdminParams.NAME, coreName,
- CommonAdminParams.WAIT_FOR_FINAL_STATE, Boolean.toString(waitForFinalState));
- Map<String, Object> propertyParams = new HashMap<>();
- ocmh.addPropertyParams(message, propertyParams);
- addReplicasProps = addReplicasProps.plus(propertyParams);
- if (async != null) addReplicasProps.getProperties().put(ASYNC, async);
- final NamedList addResult = new NamedList();
- ocmh.addReplica(zkStateReader.getClusterState(), addReplicasProps, addResult, () -> {
- countDownLatch.countDown();
- Object addResultFailure = addResult.get("failure");
- if (addResultFailure != null) {
- SimpleOrderedMap failure = (SimpleOrderedMap) results.get("failure");
- if (failure == null) {
- failure = new SimpleOrderedMap();
- results.add("failure", failure);
- }
- failure.addAll((NamedList) addResultFailure);
- } else {
- SimpleOrderedMap success = (SimpleOrderedMap) results.get("success");
- if (success == null) {
- success = new SimpleOrderedMap();
- results.add("success", success);
- }
- success.addAll((NamedList) addResult.get("success"));
- }
- });
- }
- } finally {
- if (sessionWrapper.get() != null) sessionWrapper.get().release();
- }
-
- log.debug("Waiting for create shard action to complete");
- countDownLatch.await(5, TimeUnit.MINUTES);
- log.debug("Finished waiting for create shard action to complete");
-
- log.info("Finished create command on all shards for collection: " + collectionName);
-
- }
-
- public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
- String collectionName, ZkNodeProps message, AtomicReference< PolicyHelper.SessionWrapper> sessionWrapper) throws IOException, InterruptedException {
- String sliceName = message.getStr(SHARD_ID_PROP);
- DocCollection collection = clusterState.getCollection(collectionName);
-
- int numNrtReplicas = message.getInt(NRT_REPLICAS, message.getInt(REPLICATION_FACTOR, collection.getInt(NRT_REPLICAS, collection.getInt(REPLICATION_FACTOR, 1))));
- int numPullReplicas = message.getInt(PULL_REPLICAS, collection.getInt(PULL_REPLICAS, 0));
- int numTlogReplicas = message.getInt(TLOG_REPLICAS, collection.getInt(TLOG_REPLICAS, 0));
- int totalReplicas = numNrtReplicas + numPullReplicas + numTlogReplicas;
-
- if (numNrtReplicas + numTlogReplicas <= 0) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NRT_REPLICAS + " + " + TLOG_REPLICAS + " must be greater than 0");
- }
-
- Object createNodeSetStr = message.get(OverseerCollectionMessageHandler.CREATE_NODE_SET);
-
- boolean usePolicyFramework = CloudUtil.usePolicyFramework(collection, cloudManager);
- List<ReplicaPosition> positions;
- if (usePolicyFramework) {
- if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
- positions = Assign.identifyNodes(cloudManager,
- clusterState,
- Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM),
- collection.getName(),
- message,
- Collections.singletonList(sliceName),
- numNrtReplicas,
- numTlogReplicas,
- numPullReplicas);
- sessionWrapper.set(PolicyHelper.getLastSessionWrapper(true));
- } else {
- List<Assign.ReplicaCount> sortedNodeList = getNodesForNewReplicas(clusterState, collection.getName(), sliceName, totalReplicas,
- createNodeSetStr, cloudManager);
- int i = 0;
- positions = new ArrayList<>();
- for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, numNrtReplicas,
- Replica.Type.TLOG, numTlogReplicas,
- Replica.Type.PULL, numPullReplicas
- ).entrySet()) {
- for (int j = 0; j < e.getValue(); j++) {
- positions.add(new ReplicaPosition(sliceName, j + 1, e.getKey(), sortedNodeList.get(i % sortedNodeList.size()).nodeName));
- i++;
- }
- }
- }
- return positions;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/CreateSnapshotCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateSnapshotCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateSnapshotCmd.java
deleted file mode 100644
index 5de65a4..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/CreateSnapshotCmd.java
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.solr.cloud;
-
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.CORE_NAME_PROP;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-import static org.apache.solr.common.params.CommonParams.NAME;
-
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.solr.common.SolrException;
-import org.apache.solr.common.SolrException.ErrorCode;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.Slice;
-import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.Replica.State;
-import org.apache.solr.common.params.CoreAdminParams;
-import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.core.snapshots.CollectionSnapshotMetaData;
-import org.apache.solr.core.snapshots.CollectionSnapshotMetaData.CoreSnapshotMetaData;
-import org.apache.solr.core.snapshots.CollectionSnapshotMetaData.SnapshotStatus;
-import org.apache.solr.core.snapshots.SolrSnapshotManager;
-import org.apache.solr.handler.component.ShardHandler;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This class implements the functionality of creating a collection level snapshot.
- */
-public class CreateSnapshotCmd implements OverseerCollectionMessageHandler.Cmd {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final OverseerCollectionMessageHandler ocmh;
-
- public CreateSnapshotCmd (OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- }
-
- @Override
- public void call(ClusterState state, ZkNodeProps message, NamedList results) throws Exception {
- String collectionName = message.getStr(COLLECTION_PROP);
- String commitName = message.getStr(CoreAdminParams.COMMIT_NAME);
- String asyncId = message.getStr(ASYNC);
- SolrZkClient zkClient = this.ocmh.overseer.getZkController().getZkClient();
- Date creationDate = new Date();
-
- if(SolrSnapshotManager.snapshotExists(zkClient, collectionName, commitName)) {
- throw new SolrException(ErrorCode.BAD_REQUEST, "Snapshot with name " + commitName
- + " already exists for collection " + collectionName);
- }
-
- log.info("Creating a snapshot for collection={} with commitName={}", collectionName, commitName);
-
- // Create a node in ZK to store the collection level snapshot meta-data.
- SolrSnapshotManager.createCollectionLevelSnapshot(zkClient, collectionName, new CollectionSnapshotMetaData(commitName));
- log.info("Created a ZK path to store snapshot information for collection={} with commitName={}", collectionName, commitName);
-
- Map<String, String> requestMap = new HashMap<>();
- NamedList shardRequestResults = new NamedList();
- Map<String, Slice> shardByCoreName = new HashMap<>();
- ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
-
- for (Slice slice : ocmh.zkStateReader.getClusterState().getCollection(collectionName).getSlices()) {
- for (Replica replica : slice.getReplicas()) {
- if (replica.getState() != State.ACTIVE) {
- log.info("Replica {} is not active. Hence not sending the createsnapshot request", replica.getCoreName());
- continue; // Since replica is not active - no point sending a request.
- }
-
- String coreName = replica.getStr(CORE_NAME_PROP);
-
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CoreAdminParams.ACTION, CoreAdminAction.CREATESNAPSHOT.toString());
- params.set(NAME, slice.getName());
- params.set(CORE_NAME_PROP, coreName);
- params.set(CoreAdminParams.COMMIT_NAME, commitName);
-
- ocmh.sendShardRequest(replica.getNodeName(), params, shardHandler, asyncId, requestMap);
- log.debug("Sent createsnapshot request to core={} with commitName={}", coreName, commitName);
-
- shardByCoreName.put(coreName, slice);
- }
- }
-
- // At this point we want to make sure that at-least one replica for every shard
- // is able to create the snapshot. If that is not the case, then we fail the request.
- // This is to take care of the situation where e.g. entire shard is unavailable.
- Set<String> failedShards = new HashSet<>();
-
- ocmh.processResponses(shardRequestResults, shardHandler, false, null, asyncId, requestMap);
- NamedList success = (NamedList) shardRequestResults.get("success");
- List<CoreSnapshotMetaData> replicas = new ArrayList<>();
- if (success != null) {
- for ( int i = 0 ; i < success.size() ; i++) {
- NamedList resp = (NamedList)success.getVal(i);
-
- // Check if this core is the leader for the shard. The idea here is that during the backup
- // operation we preferably use the snapshot of the "leader" replica since it is most likely
- // to have latest state.
- String coreName = (String)resp.get(CoreAdminParams.CORE);
- Slice slice = shardByCoreName.remove(coreName);
- boolean leader = (slice.getLeader() != null && slice.getLeader().getCoreName().equals(coreName));
- resp.add(SolrSnapshotManager.SHARD_ID, slice.getName());
- resp.add(SolrSnapshotManager.LEADER, leader);
-
- CoreSnapshotMetaData c = new CoreSnapshotMetaData(resp);
- replicas.add(c);
- log.info("Snapshot with commitName {} is created successfully for core {}", commitName, c.getCoreName());
- }
- }
-
- if (!shardByCoreName.isEmpty()) { // One or more failures.
- log.warn("Unable to create a snapshot with name {} for following cores {}", commitName, shardByCoreName.keySet());
-
- // Count number of failures per shard.
- Map<String, Integer> failuresByShardId = new HashMap<>();
- for (Map.Entry<String,Slice> entry : shardByCoreName.entrySet()) {
- int f = 0;
- if (failuresByShardId.get(entry.getValue().getName()) != null) {
- f = failuresByShardId.get(entry.getValue().getName());
- }
- failuresByShardId.put(entry.getValue().getName(), f + 1);
- }
-
- // Now that we know number of failures per shard, we can figure out
- // if at-least one replica per shard was able to create a snapshot or not.
- DocCollection collectionStatus = ocmh.zkStateReader.getClusterState().getCollection(collectionName);
- for (Map.Entry<String,Integer> entry : failuresByShardId.entrySet()) {
- int replicaCount = collectionStatus.getSlice(entry.getKey()).getReplicas().size();
- if (replicaCount <= entry.getValue()) {
- failedShards.add(entry.getKey());
- }
- }
- }
-
- if (failedShards.isEmpty()) { // No failures.
- CollectionSnapshotMetaData meta = new CollectionSnapshotMetaData(commitName, SnapshotStatus.Successful, creationDate, replicas);
- SolrSnapshotManager.updateCollectionLevelSnapshot(zkClient, collectionName, meta);
- log.info("Saved following snapshot information for collection={} with commitName={} in Zookeeper : {}", collectionName,
- commitName, meta.toNamedList());
- } else {
- log.warn("Failed to create a snapshot for collection {} with commitName = {}. Snapshot could not be captured for following shards {}",
- collectionName, commitName, failedShards);
- // Update the ZK meta-data to include only cores with the snapshot. This will enable users to figure out
- // which cores have the named snapshot.
- CollectionSnapshotMetaData meta = new CollectionSnapshotMetaData(commitName, SnapshotStatus.Failed, creationDate, replicas);
- SolrSnapshotManager.updateCollectionLevelSnapshot(zkClient, collectionName, meta);
- log.info("Saved following snapshot information for collection={} with commitName={} in Zookeeper : {}", collectionName,
- commitName, meta.toNamedList());
- throw new SolrException(ErrorCode.SERVER_ERROR, "Failed to create snapshot on shards " + failedShards);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/DeleteAliasCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/DeleteAliasCmd.java b/solr/core/src/java/org/apache/solr/cloud/DeleteAliasCmd.java
deleted file mode 100644
index 9c9f1c6..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/DeleteAliasCmd.java
+++ /dev/null
@@ -1,43 +0,0 @@
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.solr.cloud;
-
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.util.NamedList;
-
-import static org.apache.solr.common.params.CommonParams.NAME;
-
-public class DeleteAliasCmd implements OverseerCollectionMessageHandler.Cmd {
- private final OverseerCollectionMessageHandler ocmh;
-
- public DeleteAliasCmd(OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- }
-
- @Override
- public void call(ClusterState state, ZkNodeProps message, NamedList results) throws Exception {
- String aliasName = message.getStr(NAME);
-
- ZkStateReader zkStateReader = ocmh.zkStateReader;
- zkStateReader.aliasesHolder.applyModificationAndExportToZk(a -> a.cloneWithCollectionAlias(aliasName, null));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/DeleteCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/DeleteCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/DeleteCollectionCmd.java
deleted file mode 100644
index 8ee0168..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/DeleteCollectionCmd.java
+++ /dev/null
@@ -1,141 +0,0 @@
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.solr.cloud;
-
-import java.lang.invoke.MethodHandles;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.solr.common.NonExistentCoreException;
-import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.Aliases;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.CoreAdminParams;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.TimeSource;
-import org.apache.solr.common.util.Utils;
-import org.apache.solr.core.snapshots.SolrSnapshotManager;
-import org.apache.solr.util.TimeOut;
-import org.apache.zookeeper.KeeperException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETE;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-import static org.apache.solr.common.params.CommonParams.NAME;
-
-public class DeleteCollectionCmd implements OverseerCollectionMessageHandler.Cmd {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final OverseerCollectionMessageHandler ocmh;
- private final TimeSource timeSource;
-
- public DeleteCollectionCmd(OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- this.timeSource = ocmh.cloudManager.getTimeSource();
- }
-
- @Override
- public void call(ClusterState state, ZkNodeProps message, NamedList results) throws Exception {
- ZkStateReader zkStateReader = ocmh.zkStateReader;
- Aliases aliases = zkStateReader.getAliases();
- final String collection = message.getStr(NAME);
- for (Map.Entry<String, List<String>> ent : aliases.getCollectionAliasListMap().entrySet()) {
- if (ent.getValue().contains(collection)) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
- "Collection : " + collection + " is part of alias " + ent.getKey() + " remove or modify the alias before removing this collection.");
- }
- }
-
- try {
- // Remove the snapshots meta-data for this collection in ZK. Deleting actual index files
- // should be taken care of as part of collection delete operation.
- SolrZkClient zkClient = zkStateReader.getZkClient();
- SolrSnapshotManager.cleanupCollectionLevelSnapshots(zkClient, collection);
-
- if (zkStateReader.getClusterState().getCollectionOrNull(collection) == null) {
- if (zkStateReader.getZkClient().exists(ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection, true)) {
- // if the collection is not in the clusterstate, but is listed in zk, do nothing, it will just
- // be removed in the finally - we cannot continue, because the below code will error if the collection
- // is not in the clusterstate
- return;
- }
- }
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.UNLOAD.toString());
- params.set(CoreAdminParams.DELETE_INSTANCE_DIR, true);
- params.set(CoreAdminParams.DELETE_DATA_DIR, true);
-
- String asyncId = message.getStr(ASYNC);
- Map<String, String> requestMap = null;
- if (asyncId != null) {
- requestMap = new HashMap<>();
- }
-
- Set<String> okayExceptions = new HashSet<>(1);
- okayExceptions.add(NonExistentCoreException.class.getName());
-
- ocmh.collectionCmd(message, params, results, null, asyncId, requestMap, okayExceptions);
-
- ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, DELETE.toLower(), NAME, collection);
- Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(m));
-
- // wait for a while until we don't see the collection
- TimeOut timeout = new TimeOut(30, TimeUnit.SECONDS, timeSource);
- boolean removed = false;
- while (! timeout.hasTimedOut()) {
- timeout.sleep(100);
- removed = !zkStateReader.getClusterState().hasCollection(collection);
- if (removed) {
- timeout.sleep(500); // just a bit of time so it's more likely other
- // readers see on return
- break;
- }
- }
- if (!removed) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
- "Could not fully remove collection: " + collection);
- }
-
- } finally {
-
- try {
- if (zkStateReader.getZkClient().exists(
- ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection, true)) {
- zkStateReader.getZkClient().clean(
- ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection);
- }
- } catch (InterruptedException e) {
- SolrException.log(log, "Cleaning up collection in zk was interrupted:"
- + collection, e);
- Thread.currentThread().interrupt();
- } catch (KeeperException e) {
- SolrException.log(log, "Problem cleaning up collection in zk:"
- + collection, e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/DeleteNodeCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/DeleteNodeCmd.java b/solr/core/src/java/org/apache/solr/cloud/DeleteNodeCmd.java
deleted file mode 100644
index 51b0956..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/DeleteNodeCmd.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.solr.cloud;
-
-
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Locale;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-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.util.NamedList;
-import org.apache.zookeeper.KeeperException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEREPLICA;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-
-public class DeleteNodeCmd implements OverseerCollectionMessageHandler.Cmd {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- private final OverseerCollectionMessageHandler ocmh;
-
- public DeleteNodeCmd(OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- }
-
- @Override
- public void call(ClusterState state, ZkNodeProps message, NamedList results) throws Exception {
- ocmh.checkRequired(message, "node");
- String node = message.getStr("node");
- if (!state.liveNodesContain(node)) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Source Node: " + node + " is not live");
- }
- List<ZkNodeProps> sourceReplicas = ReplaceNodeCmd.getReplicasOfNode(node, state);
- List<String> singleReplicas = verifyReplicaAvailability(sourceReplicas, state);
- if (!singleReplicas.isEmpty()) {
- results.add("failure", "Can't delete the only existing non-PULL replica(s) on node " + node + ": " + singleReplicas.toString());
- } else {
- cleanupReplicas(results, state, sourceReplicas, ocmh, node, message.getStr(ASYNC));
- }
- }
-
- // collect names of replicas that cannot be deleted
- static List<String> verifyReplicaAvailability(List<ZkNodeProps> sourceReplicas, ClusterState state) {
- List<String> res = new ArrayList<>();
- for (ZkNodeProps sourceReplica : sourceReplicas) {
- String coll = sourceReplica.getStr(COLLECTION_PROP);
- String shard = sourceReplica.getStr(SHARD_ID_PROP);
- String replicaName = sourceReplica.getStr(ZkStateReader.REPLICA_PROP);
- DocCollection collection = state.getCollection(coll);
- Slice slice = collection.getSlice(shard);
- if (slice.getReplicas().size() < 2) {
- // can't delete the only replica in existence
- res.add(coll + "/" + shard + "/" + replicaName + ", type=" + sourceReplica.getStr(ZkStateReader.REPLICA_TYPE));
- } else { // check replica types
- int otherNonPullReplicas = 0;
- for (Replica r : slice.getReplicas()) {
- if (!r.getName().equals(replicaName) && !r.getType().equals(Replica.Type.PULL)) {
- otherNonPullReplicas++;
- }
- }
- // can't delete - there are no other non-pull replicas
- if (otherNonPullReplicas == 0) {
- res.add(coll + "/" + shard + "/" + replicaName + ", type=" + sourceReplica.getStr(ZkStateReader.REPLICA_TYPE));
- }
- }
- }
- return res;
- }
-
- static void cleanupReplicas(NamedList results,
- ClusterState clusterState,
- List<ZkNodeProps> sourceReplicas,
- OverseerCollectionMessageHandler ocmh,
- String node,
- String async) throws InterruptedException {
- CountDownLatch cleanupLatch = new CountDownLatch(sourceReplicas.size());
- for (ZkNodeProps sourceReplica : sourceReplicas) {
- String coll = sourceReplica.getStr(COLLECTION_PROP);
- String shard = sourceReplica.getStr(SHARD_ID_PROP);
- String type = sourceReplica.getStr(ZkStateReader.REPLICA_TYPE);
- log.info("Deleting replica type={} for collection={} shard={} on node={}", type, coll, shard, node);
- NamedList deleteResult = new NamedList();
- try {
- if (async != null) sourceReplica = sourceReplica.plus(ASYNC, async);
- ((DeleteReplicaCmd)ocmh.commandMap.get(DELETEREPLICA)).deleteReplica(clusterState, sourceReplica.plus("parallel", "true"), deleteResult, () -> {
- cleanupLatch.countDown();
- if (deleteResult.get("failure") != null) {
- synchronized (results) {
-
- results.add("failure", String.format(Locale.ROOT, "Failed to delete replica for collection=%s shard=%s" +
- " on node=%s", coll, shard, node));
- }
- }
- });
- } catch (KeeperException e) {
- log.warn("Error deleting ", e);
- cleanupLatch.countDown();
- } catch (Exception e) {
- log.warn("Error deleting ", e);
- cleanupLatch.countDown();
- throw e;
- }
- }
- log.debug("Waiting for delete node action to complete");
- cleanupLatch.await(5, TimeUnit.MINUTES);
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/DeleteReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/DeleteReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/DeleteReplicaCmd.java
deleted file mode 100644
index e71d7e8..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/DeleteReplicaCmd.java
+++ /dev/null
@@ -1,281 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.solr.cloud;
-
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
-import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-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.CoreAdminParams;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.StrUtils;
-import org.apache.solr.common.util.Utils;
-import org.apache.solr.handler.component.ShardHandler;
-import org.apache.zookeeper.KeeperException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.ONLY_IF_DOWN;
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
-import static org.apache.solr.common.params.CollectionAdminParams.COUNT_PROP;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-
-
-public class DeleteReplicaCmd implements Cmd {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final OverseerCollectionMessageHandler ocmh;
-
- public DeleteReplicaCmd(OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- }
-
- @Override
- @SuppressWarnings("unchecked")
-
- public void call(ClusterState clusterState, ZkNodeProps message, NamedList results) throws Exception {
- deleteReplica(clusterState, message, results,null);
- }
-
-
- @SuppressWarnings("unchecked")
- void deleteReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
- throws KeeperException, InterruptedException {
- log.debug("deleteReplica() : {}", Utils.toJSONString(message));
- boolean parallel = message.getBool("parallel", false);
-
- //If a count is specified the strategy needs be different
- if (message.getStr(COUNT_PROP) != null) {
- deleteReplicaBasedOnCount(clusterState, message, results, onComplete, parallel);
- return;
- }
-
-
- ocmh.checkRequired(message, COLLECTION_PROP, SHARD_ID_PROP, REPLICA_PROP);
- String collectionName = message.getStr(COLLECTION_PROP);
- String shard = message.getStr(SHARD_ID_PROP);
- String replicaName = message.getStr(REPLICA_PROP);
-
- DocCollection coll = clusterState.getCollection(collectionName);
- Slice slice = coll.getSlice(shard);
- if (slice == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
- "Invalid shard name : " + shard + " in collection : " + collectionName);
- }
-
- deleteCore(slice, collectionName, replicaName, message, shard, results, onComplete, parallel);
-
- }
-
-
- /**
- * Delete replicas based on count for a given collection. If a shard is passed, uses that
- * else deletes given num replicas across all shards for the given collection.
- */
- void deleteReplicaBasedOnCount(ClusterState clusterState,
- ZkNodeProps message,
- NamedList results,
- Runnable onComplete,
- boolean parallel)
- throws KeeperException, InterruptedException {
- ocmh.checkRequired(message, COLLECTION_PROP, COUNT_PROP);
- int count = Integer.parseInt(message.getStr(COUNT_PROP));
- String collectionName = message.getStr(COLLECTION_PROP);
- String shard = message.getStr(SHARD_ID_PROP);
- DocCollection coll = clusterState.getCollection(collectionName);
- Slice slice = null;
- //Validate if shard is passed.
- if (shard != null) {
- slice = coll.getSlice(shard);
- if (slice == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
- "Invalid shard name : " + shard + " in collection : " + collectionName);
- }
- }
-
- Map<Slice, Set<String>> shardToReplicasMapping = new HashMap<Slice, Set<String>>();
- if (slice != null) {
- Set<String> replicasToBeDeleted = pickReplicasTobeDeleted(slice, shard, collectionName, count);
- shardToReplicasMapping.put(slice,replicasToBeDeleted);
- } else {
-
- //If there are many replicas left, remove the rest based on count.
- Collection<Slice> allSlices = coll.getSlices();
- for (Slice individualSlice : allSlices) {
- Set<String> replicasToBeDeleted = pickReplicasTobeDeleted(individualSlice, individualSlice.getName(), collectionName, count);
- shardToReplicasMapping.put(individualSlice, replicasToBeDeleted);
- }
- }
-
- for (Slice shardSlice: shardToReplicasMapping.keySet()) {
- String shardId = shardSlice.getName();
- Set<String> replicas = shardToReplicasMapping.get(shardSlice);
- //callDeleteReplica on all replicas
- for (String replica: replicas) {
- log.debug("Deleting replica {} for shard {} based on count {}", replica, shardId, count);
- deleteCore(shardSlice, collectionName, replica, message, shard, results, onComplete, parallel);
- }
- results.add("shard_id", shardId);
- results.add("replicas_deleted", replicas);
- }
-
- }
-
-
- /**
- * Pick replicas to be deleted. Avoid picking the leader.
- */
- private Set<String> pickReplicasTobeDeleted(Slice slice, String shard, String collectionName, int count) {
- validateReplicaAvailability(slice, shard, collectionName, count);
- Collection<Replica> allReplicas = slice.getReplicas();
- Set<String> replicasToBeRemoved = new HashSet<String>();
- Replica leader = slice.getLeader();
- for (Replica replica: allReplicas) {
- if (count == 0) {
- break;
- }
- //Try avoiding to pick up the leader to minimize activity on the cluster.
- if (leader.getCoreName().equals(replica.getCoreName())) {
- continue;
- }
- replicasToBeRemoved.add(replica.getName());
- count --;
- }
- return replicasToBeRemoved;
- }
-
- /**
- * Validate if there is less replicas than requested to remove. Also error out if there is
- * only one replica available
- */
- private void validateReplicaAvailability(Slice slice, String shard, String collectionName, int count) {
- //If there is a specific shard passed, validate if there any or just 1 replica left
- if (slice != null) {
- Collection<Replica> allReplicasForShard = slice.getReplicas();
- if (allReplicasForShard == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No replicas found in shard/collection: " +
- shard + "/" + collectionName);
- }
-
-
- if (allReplicasForShard.size() == 1) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "There is only one replica available in shard/collection: " +
- shard + "/" + collectionName + ". Cannot delete that.");
- }
-
- if (allReplicasForShard.size() <= count) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "There are lesser num replicas requested to be deleted than are available in shard/collection : " +
- shard + "/" + collectionName + " Requested: " + count + " Available: " + allReplicasForShard.size() + ".");
- }
- }
- }
-
- void deleteCore(Slice slice, String collectionName, String replicaName,ZkNodeProps message, String shard, NamedList results, Runnable onComplete, boolean parallel) throws KeeperException, InterruptedException {
-
- Replica replica = slice.getReplica(replicaName);
- if (replica == null) {
- ArrayList<String> l = new ArrayList<>();
- for (Replica r : slice.getReplicas())
- l.add(r.getName());
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Invalid replica : " + replicaName + " in shard/collection : " +
- shard + "/" + collectionName + " available replicas are " + StrUtils.join(l, ','));
- }
-
- // If users are being safe and only want to remove a shard if it is down, they can specify onlyIfDown=true
- // on the command.
- if (Boolean.parseBoolean(message.getStr(ONLY_IF_DOWN)) && replica.getState() != Replica.State.DOWN) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
- "Attempted to remove replica : " + collectionName + "/" + shard + "/" + replicaName +
- " with onlyIfDown='true', but state is '" + replica.getStr(ZkStateReader.STATE_PROP) + "'");
- }
-
- ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
- String core = replica.getStr(ZkStateReader.CORE_NAME_PROP);
- String asyncId = message.getStr(ASYNC);
- AtomicReference<Map<String, String>> requestMap = new AtomicReference<>(null);
- if (asyncId != null) {
- requestMap.set(new HashMap<>(1, 1.0f));
- }
-
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.add(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.UNLOAD.toString());
- params.add(CoreAdminParams.CORE, core);
-
- params.set(CoreAdminParams.DELETE_INDEX, message.getBool(CoreAdminParams.DELETE_INDEX, true));
- params.set(CoreAdminParams.DELETE_INSTANCE_DIR, message.getBool(CoreAdminParams.DELETE_INSTANCE_DIR, true));
- params.set(CoreAdminParams.DELETE_DATA_DIR, message.getBool(CoreAdminParams.DELETE_DATA_DIR, true));
-
- boolean isLive = ocmh.zkStateReader.getClusterState().getLiveNodes().contains(replica.getNodeName());
- if (isLive) {
- ocmh.sendShardRequest(replica.getNodeName(), params, shardHandler, asyncId, requestMap.get());
- }
-
- Callable<Boolean> callable = () -> {
- try {
- if (isLive) {
- ocmh.processResponses(results, shardHandler, false, null, asyncId, requestMap.get());
-
- //check if the core unload removed the corenode zk entry
- if (ocmh.waitForCoreNodeGone(collectionName, shard, replicaName, 5000)) return Boolean.TRUE;
- }
-
- // try and ensure core info is removed from cluster state
- ocmh.deleteCoreNode(collectionName, replicaName, replica, core);
- if (ocmh.waitForCoreNodeGone(collectionName, shard, replicaName, 30000)) return Boolean.TRUE;
- return Boolean.FALSE;
- } catch (Exception e) {
- results.add("failure", "Could not complete delete " + e.getMessage());
- throw e;
- } finally {
- if (onComplete != null) onComplete.run();
- }
- };
-
- if (!parallel) {
- try {
- if (!callable.call())
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
- "Could not remove replica : " + collectionName + "/" + shard + "/" + replicaName);
- } catch (InterruptedException | KeeperException e) {
- throw e;
- } catch (Exception ex) {
- throw new SolrException(SolrException.ErrorCode.UNKNOWN, "Error waiting for corenode gone", ex);
- }
-
- } else {
- ocmh.tpe.submit(callable);
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a3c4f738/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java
deleted file mode 100644
index 58c4e63..0000000
--- a/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java
+++ /dev/null
@@ -1,178 +0,0 @@
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.solr.cloud;
-
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.solr.client.solrj.cloud.DistributedQueue;
-import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
-import org.apache.solr.cloud.overseer.OverseerAction;
-import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-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.CoreAdminParams;
-import org.apache.solr.common.util.NamedList;
-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.util.TimeOut;
-import org.apache.zookeeper.KeeperException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.NODE_NAME_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEREPLICA;
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETESHARD;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-
-public class DeleteShardCmd implements Cmd {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final OverseerCollectionMessageHandler ocmh;
- private final TimeSource timeSource;
-
- public DeleteShardCmd(OverseerCollectionMessageHandler ocmh) {
- this.ocmh = ocmh;
- this.timeSource = ocmh.cloudManager.getTimeSource();
- }
-
- @Override
- public void call(ClusterState clusterState, ZkNodeProps message, NamedList results) throws Exception {
- String collectionName = message.getStr(ZkStateReader.COLLECTION_PROP);
- String sliceId = message.getStr(ZkStateReader.SHARD_ID_PROP);
-
- log.info("Delete shard invoked");
- Slice slice = clusterState.getCollection(collectionName).getSlice(sliceId);
- if (slice == null) throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
- "No shard with name " + sliceId + " exists for collection " + collectionName);
-
- // For now, only allow for deletions of Inactive slices or custom hashes (range==null).
- // TODO: Add check for range gaps on Slice deletion
- final Slice.State state = slice.getState();
- if (!(slice.getRange() == null || state == Slice.State.INACTIVE || state == Slice.State.RECOVERY
- || state == Slice.State.CONSTRUCTION) || state == Slice.State.RECOVERY_FAILED) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "The slice: " + slice.getName() + " is currently " + state
- + ". Only non-active (or custom-hashed) slices can be deleted.");
- }
-
- if (state == Slice.State.RECOVERY) {
- // mark the slice as 'construction' and only then try to delete the cores
- // see SOLR-9455
- DistributedQueue inQueue = Overseer.getStateUpdateQueue(ocmh.zkStateReader.getZkClient());
- Map<String, Object> propMap = new HashMap<>();
- propMap.put(Overseer.QUEUE_OPERATION, OverseerAction.UPDATESHARDSTATE.toLower());
- propMap.put(sliceId, Slice.State.CONSTRUCTION.toString());
- propMap.put(ZkStateReader.COLLECTION_PROP, collectionName);
- ZkNodeProps m = new ZkNodeProps(propMap);
- inQueue.offer(Utils.toJSON(m));
- }
-
- String asyncId = message.getStr(ASYNC);
-
- try {
- List<ZkNodeProps> replicas = getReplicasForSlice(collectionName, slice);
- CountDownLatch cleanupLatch = new CountDownLatch(replicas.size());
- for (ZkNodeProps r : replicas) {
- final ZkNodeProps replica = r.plus(message.getProperties()).plus("parallel", "true").plus(ASYNC, asyncId);
- log.info("Deleting replica for collection={} shard={} on node={}", replica.getStr(COLLECTION_PROP), replica.getStr(SHARD_ID_PROP), replica.getStr(CoreAdminParams.NODE));
- NamedList deleteResult = new NamedList();
- try {
- ((DeleteReplicaCmd)ocmh.commandMap.get(DELETEREPLICA)).deleteReplica(clusterState, replica, deleteResult, () -> {
- cleanupLatch.countDown();
- if (deleteResult.get("failure") != null) {
- synchronized (results) {
- results.add("failure", String.format(Locale.ROOT, "Failed to delete replica for collection=%s shard=%s" +
- " on node=%s", replica.getStr(COLLECTION_PROP), replica.getStr(SHARD_ID_PROP), replica.getStr(NODE_NAME_PROP)));
- }
- }
- SimpleOrderedMap success = (SimpleOrderedMap) deleteResult.get("success");
- if (success != null) {
- synchronized (results) {
- results.add("success", success);
- }
- }
- });
- } catch (KeeperException e) {
- log.warn("Error deleting replica: " + r, e);
- cleanupLatch.countDown();
- } catch (Exception e) {
- log.warn("Error deleting replica: " + r, e);
- cleanupLatch.countDown();
- throw e;
- }
- }
- log.debug("Waiting for delete shard action to complete");
- cleanupLatch.await(5, TimeUnit.MINUTES);
-
- ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, DELETESHARD.toLower(), ZkStateReader.COLLECTION_PROP,
- collectionName, ZkStateReader.SHARD_ID_PROP, sliceId);
- ZkStateReader zkStateReader = ocmh.zkStateReader;
- Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(m));
-
- // wait for a while until we don't see the shard
- TimeOut timeout = new TimeOut(30, TimeUnit.SECONDS, timeSource);
- boolean removed = false;
- while (!timeout.hasTimedOut()) {
- timeout.sleep(100);
- DocCollection collection = zkStateReader.getClusterState().getCollection(collectionName);
- removed = collection.getSlice(sliceId) == null;
- if (removed) {
- timeout.sleep(100); // just a bit of time so it's more likely other readers see on return
- break;
- }
- }
- if (!removed) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
- "Could not fully remove collection: " + collectionName + " shard: " + sliceId);
- }
-
- log.info("Successfully deleted collection: " + collectionName + ", shard: " + sliceId);
- } catch (SolrException e) {
- throw e;
- } catch (Exception e) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
- "Error executing delete operation for collection: " + collectionName + " shard: " + sliceId, e);
- }
- }
-
- private List<ZkNodeProps> getReplicasForSlice(String collectionName, Slice slice) {
- List<ZkNodeProps> sourceReplicas = new ArrayList<>();
- for (Replica replica : slice.getReplicas()) {
- ZkNodeProps props = new ZkNodeProps(
- COLLECTION_PROP, collectionName,
- SHARD_ID_PROP, slice.getName(),
- ZkStateReader.CORE_NAME_PROP, replica.getCoreName(),
- ZkStateReader.REPLICA_PROP, replica.getName(),
- CoreAdminParams.NODE, replica.getNodeName());
- sourceReplicas.add(props);
- }
- return sourceReplicas;
- }
-}