How Solr Create Collection - Learn Solr Code

Test Code to create collections
MiniSolrCloudCluster cluster = new MiniSolrCloudCluster(4 /*numServers*/, testBaseDir, solrXml, JettyConfig.builder().setContext("/solr").build());
cluster.createCollection(collectionName, 2/*numShards*/, 2/*replicationFactor*/, "cie-default", null);

CollectionsHandler

CollectionsHandler.handleRequestBody(SolrQueryRequest, SolrQueryResponse)

CollectionAction action = CollectionAction.get(a); // CollectionAction .CREATE(true)

CollectionOperation operation = CollectionOperation.get(action); //CollectionOperation .CREATE_OP(CREATE)

Map result = operation.call(req, rsp, this);

Return a mpa like this:

{name=collectionName, fromApi=true, replicationFactor=2, collection.configName=configName, numShards=2, stateFormat=2}

ZkNodeProps props = new ZkNodeProps(result);

if (operation.sendToOCPQueue) handleResponse(operation.action.toLower(), props, rsp, operation.timeOut);

CollectionsHandler. handleResponse

QueueEvent event = coreContainer.getZkController() .getOverseerCollectionQueue() .offer(Utils.toJSON(m), timeout);

This uses DistributedQueue.offer(byte[] data, long timeout) to add a task to /overseer/collection-queue-work/qnr-numbers.

It uses LatchWatcher to wait until this task is processed.

Overseer and OverseerCollectionProcessor

OverseerCollectionProcessor.processMessage(ZkNodeProps, String operation /*create*/)

OverseerCollectionProcessor.processMessage(ZkNodeProps, String)

{

  "name":"collectionName",

  "fromApi":"true",

  "replicationFactor":"2",

  "collection.configName":"configName",

  "numShards":"2",

  "stateFormat":"2",

  "operation":"create"}

  

OverseerCollectionProcessor.createCollection(ClusterState, ZkNodeProps, NamedList)

  ClusterStateMutator.getShardNames(numSlices, shardNames);

   positionVsNodes = identifyNodes(clusterState, nodeList, message, shardNames, repFactor); // round-robin if rule not set

  createConfNode(configName, collectionName, isLegacyCloud);

    Overseer.getInQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(message));

// This message will be processed by ClusterStateUpdater

// wait for a while until we do see the collection in the clusterState

  for (Map.Entry e : positionVsNodes.entrySet()) {

  if (isLegacyCloud) {

    shardHandler.submit(sreq, sreq.shards[0], sreq.params);

  } else {

    coresToCreate.put(coreName, sreq);

  }

}

This will send http call and be handled by CoreAdminHandler.handleRequestBody.

ClusterStateUpdater

  // if there were any errors while processing

  // the state queue, items would have been left in the

  // work queue so let's process those first

  byte[] data = workQueue.peek();

  boolean hadWorkItems = data != null;

  while (data != null)  {

    final ZkNodeProps message = ZkNodeProps.load(data);

    clusterState = processQueueItem(message, clusterState, zkStateWriter, false, null);

    workQueue.poll(); // poll-ing removes the element we got by peek-ing

    data = workQueue.peek();

  }

ClusterStateUpdater.processQueueItem

  zkWriteCommand = processMessage(clusterState, message, operation);

  stats.success(operation);

  clusterState = zkStateWriter.enqueueUpdate(clusterState, zkWriteCommand, callback);

processMessage

  case CREATE:

    return new ClusterStateMutator(getZkStateReader()).createCollection(clusterState, message);

overseer.ClusterStateMutator.createCollection(ClusterState, ZkNodeProps)