import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
protected static String printClusterStateInfo(String collection) throws Exception {
  cloudSolrClient.getZkStateReader().updateClusterState();
  String cs = null;
  ClusterState clusterState = cloudSolrClient.getZkStateReader().getClusterState();
  if (collection != null) {
    cs = clusterState.getCollection(collection).toString();
  } else {
    Map<String, DocCollection> map = new HashMap<String, DocCollection>();
    for (String coll : clusterState.getCollections())
      map.put(coll, clusterState.getCollection(coll));
    CharArr out = new CharArr();
    new JSONWriter(out, 2).write(map);
    cs = out.toString();
  }
  return cs;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
protected String printClusterStateInfo(String collection) throws Exception {
  cloudClient.getZkStateReader().updateClusterState(true);
  String cs = null;
  ClusterState clusterState = cloudClient.getZkStateReader().getClusterState();
  if (collection != null) {
    cs = clusterState.getCollection(collection).toString();
  } else {
    Map<String,DocCollection> map = new HashMap<String,DocCollection>();
    for (String coll : clusterState.getCollections())
      map.put(coll, clusterState.getCollection(coll));
    CharArr out = new CharArr();
    new JSONWriter(out, 2).write(map);
    cs = out.toString();
  }
  return cs;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
public List<List<String>> extractShardUrls(String zkHost, String collection) {

    DocCollection docCollection = extractDocCollection(zkHost, collection);
    List<Slice> slices = getSortedSlices(docCollection.getSlices());
    List<List<String>> solrUrls = new ArrayList<>(slices.size());
    for (Slice slice : slices) {
      if (slice.getLeader() == null) {
        throw new IllegalArgumentException("Cannot find SolrCloud slice leader. " +
            "It looks like not all of your shards are registered in ZooKeeper yet");
      }
      Collection<Replica> replicas = slice.getReplicas();
      List<String> urls = new ArrayList<>(replicas.size());
      for (Replica replica : replicas) {
        ZkCoreNodeProps props = new ZkCoreNodeProps(replica);
        urls.add(props.getCoreUrl());
      }
      solrUrls.add(urls);
    }
    return solrUrls;
  }
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private boolean amISubShardLeader(DocCollection coll, Slice parentSlice, String id, SolrInputDocument doc) throws InterruptedException {
  // Am I the leader of a shard in "construction/recovery" state?
  String myShardId = req.getCore().getCoreDescriptor().getCloudDescriptor().getShardId();
  Slice mySlice = coll.getSlice(myShardId);
  String state = mySlice.getState();
  if (Slice.CONSTRUCTION.equals(state) || Slice.RECOVERY.equals(state)) {
    Replica myLeader = zkController.getZkStateReader().getLeaderRetry(collection, myShardId);
    boolean amILeader = myLeader.getName().equals(
        req.getCore().getCoreDescriptor().getCloudDescriptor()
            .getCoreNodeName());
    if (amILeader) {
      // Does the document belong to my hash range as well?
      DocRouter.Range myRange = mySlice.getRange();
      if (myRange == null) myRange = new DocRouter.Range(Integer.MIN_VALUE, Integer.MAX_VALUE);
      if (parentSlice != null)  {
        boolean isSubset = parentSlice.getRange() != null && myRange.isSubsetOf(parentSlice.getRange());
        return isSubset && coll.getRouter().isTargetSlice(id, doc, req.getParams(), myShardId, coll);
      } else  {
        // delete by query case -- as long as I am a sub shard leader we're fine
        return true;
      }
    }
  }
  return false;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private List<Node> getSubShardLeaders(DocCollection coll, String shardId, String docId, SolrInputDocument doc) {
  Collection<Slice> allSlices = coll.getSlices();
  List<Node> nodes = null;
  for (Slice aslice : allSlices) {
    if (Slice.CONSTRUCTION.equals(aslice.getState()) || Slice.RECOVERY.equals(aslice.getState()))  {
      DocRouter.Range myRange = coll.getSlice(shardId).getRange();
      if (myRange == null) myRange = new DocRouter.Range(Integer.MIN_VALUE, Integer.MAX_VALUE);
      boolean isSubset = aslice.getRange() != null && aslice.getRange().isSubsetOf(myRange);
      if (isSubset &&
          (docId == null // in case of deletes
          || (docId != null && coll.getRouter().isTargetSlice(docId, doc, req.getParams(), aslice.getName(), coll)))) {
        Replica sliceLeader = aslice.getLeader();
        // slice leader can be null because node/shard is created zk before leader election
        if (sliceLeader != null && zkController.getClusterState().liveNodesContain(sliceLeader.getNodeName()))  {
          if (nodes == null) nodes = new ArrayList<>();
          ZkCoreNodeProps nodeProps = new ZkCoreNodeProps(sliceLeader);
          nodes.add(new StdNode(nodeProps, coll.getName(), shardId));
        }
      }
    }
  }
  return nodes;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private boolean checkIfCoreNodeNameAlreadyExists(CoreDescriptor dcore) {
  ZkStateReader zkStateReader = coreContainer.getZkController()
      .getZkStateReader();
  DocCollection collection = zkStateReader.getClusterState().getCollectionOrNull(dcore.getCollectionName());
  if (collection != null) {
    Collection<Slice> slices = collection.getSlices();
    
    for (Slice slice : slices) {
      Collection<Replica> replicas = slice.getReplicas();
      for (Replica replica : replicas) {
        if (replica.getName().equals(
            dcore.getCloudDescriptor().getCoreNodeName())) {
          return true;
        }
      }
    }
  }
  return false;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private boolean waitForCoreNodeGone(String collectionName, String shard, String replicaName, int timeoutms) throws InterruptedException {
  long waitUntil = System.nanoTime() + TimeUnit.NANOSECONDS.convert(timeoutms, TimeUnit.MILLISECONDS);
  boolean deleted = false;
  while (System.nanoTime() < waitUntil) {
    Thread.sleep(100);
    DocCollection docCollection = zkStateReader.getClusterState().getCollection(collectionName);
    if(docCollection != null) {
      Slice slice = docCollection.getSlice(shard);
      if(slice == null || slice.getReplica(replicaName) == null) {
        deleted =  true;
      }
    }
    // Return true if either someone already deleted the collection/slice/replica.
    if (docCollection == null || deleted) break;
  }
  return deleted;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private void waitForNewShard(String collectionName, String sliceName) throws KeeperException, InterruptedException {
  log.info("Waiting for slice {} of collection {} to be available", sliceName, collectionName);
  long startTime = System.currentTimeMillis();
  int retryCount = 320;
  while (retryCount-- > 0) {
    DocCollection collection = zkStateReader.getClusterState().getCollection(collectionName);
    if (collection == null) {
      throw new SolrException(ErrorCode.SERVER_ERROR,
          "Unable to find collection: " + collectionName + " in clusterstate");
    }
    Slice slice = collection.getSlice(sliceName);
    if (slice != null) {
      log.info("Waited for {} seconds for slice {} of collection {} to be available",
          (System.currentTimeMillis() - startTime) / 1000, sliceName, collectionName);
      return;
    }
    Thread.sleep(1000);
    zkStateReader.updateClusterState(true);
  }
  throw new SolrException(ErrorCode.SERVER_ERROR,
      "Could not find new slice " + sliceName + " in collection " + collectionName
          + " even after waiting for " + (System.currentTimeMillis() - startTime) / 1000 + " seconds"
  );
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
@Test
public void testDefaultSliceState() throws Exception {
  Map<String, DocCollection> collectionStates = new HashMap<>();
  Set<String> liveNodes = new HashSet<>();
  liveNodes.add("node1");

  Map<String, Slice> slices = new HashMap<>();
  Map<String, Replica> sliceToProps = new HashMap<>();
  Map<String, Object> props = new HashMap<>();

  Replica replica = new Replica("node1", props);
  sliceToProps.put("node1", replica);
  Slice slice = new Slice("shard1", sliceToProps, null);
  assertEquals("Default state not set to active", Slice.ACTIVE, slice.getState());
  slices.put("shard1", slice);
  collectionStates.put("collection1", new DocCollection("collection1", slices, null, DocRouter.DEFAULT));

  ZkStateReader mockZkStateReader = ClusterStateTest.getMockZkStateReader(collectionStates.keySet());
  ClusterState clusterState = new ClusterState(-1,liveNodes, collectionStates);
  byte[] bytes = ZkStateReader.toJSON(clusterState);
  ClusterState loadedClusterState = ClusterState.load(-1, bytes, liveNodes);

  assertEquals("Default state not set to active", "active", loadedClusterState.getSlice("collection1", "shard1").getState());
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private void waitTillRecovered() throws Exception {
  for (int i = 0; i < 30; i++) {
    Thread.sleep(3000);
    ZkStateReader zkStateReader = cloudClient.getZkStateReader();
    zkStateReader.updateClusterState(true);
    ClusterState clusterState = zkStateReader.getClusterState();
    DocCollection collection1 = clusterState.getCollection("collection1");
    Slice slice = collection1.getSlice("shard1");
    Collection<Replica> replicas = slice.getReplicas();
    boolean allActive = true;
    for (Replica replica : replicas) {
      if (!clusterState.liveNodesContain(replica.getNodeName())
          || !replica.get(ZkStateReader.STATE_PROP).equals(
          ZkStateReader.ACTIVE)) {
        allActive = false;
        break;
      }
    }
    if (allActive) {
      return;
    }
  }
  printLayout();
  fail("timeout waiting to see recovered node");
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private void waitTillAllNodesActive() throws Exception {
  for (int i = 0; i < 60; i++) { 
    Thread.sleep(3000);
    ZkStateReader zkStateReader = cloudClient.getZkStateReader();
    zkStateReader.updateClusterState(true);
    ClusterState clusterState = zkStateReader.getClusterState();
    DocCollection collection1 = clusterState.getCollection("collection1");
    Slice slice = collection1.getSlice("shard1");
    Collection<Replica> replicas = slice.getReplicas();
    boolean allActive = true;
    for (Replica replica : replicas) {
      if (!clusterState.liveNodesContain(replica.getNodeName())
          || !replica.get(ZkStateReader.STATE_PROP).equals(
              ZkStateReader.ACTIVE)) {
        allActive = false;
        break;
      }
    }
    if (allActive) {
      return;
    }
  }
  printLayout();
  fail("timeout waiting to see all nodes active");
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
protected void removeAndWaitForReplicaGone(String COLL_NAME,
    CloudSolrServer client, Replica replica, String shard)
    throws SolrServerException, IOException, InterruptedException {
  Map m = makeMap("collection", COLL_NAME, "action", DELETEREPLICA, "shard",
      shard, "replica", replica.getName());
  SolrParams params = new MapSolrParams(m);
  SolrRequest request = new QueryRequest(params);
  request.setPath("/admin/collections");
  client.request(request);
  long endAt = System.currentTimeMillis() + 3000;
  boolean success = false;
  DocCollection testcoll = null;
  while (System.currentTimeMillis() < endAt) {
    testcoll = getCommonCloudSolrServer().getZkStateReader()
        .getClusterState().getCollection(COLL_NAME);
    success = testcoll.getSlice(shard).getReplica(replica.getName()) == null;
    if (success) {
      log.info("replica cleaned up {}/{} core {}",
          shard + "/" + replica.getName(), replica.getStr("core"));
      log.info("current state {}", testcoll);
      break;
    }
    Thread.sleep(100);
  }
  assertTrue("Replica not cleaned up", success);
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
@Override
public void doTest() throws Exception {
  int replicationFactor = 1;
  int maxShardsPerNode = 5;

  Map<String, Object> props = ZkNodeProps.makeMap(
      "router.name", ImplicitDocRouter.NAME,
      REPLICATION_FACTOR, replicationFactor,
      MAX_SHARDS_PER_NODE, maxShardsPerNode,
      NUM_SLICES, 1,
      SHARDS_PROP,"a,b");

  Map<String,List<Integer>> collectionInfos = new HashMap<>();

  String collectionName = "customcollreplicadeletion";

  createCollection(collectionInfos, collectionName, props, client);

  waitForRecoveriesToFinish(collectionName, false);

  DocCollection testcoll = getCommonCloudSolrServer().getZkStateReader()
      .getClusterState().getCollection(collectionName);
  Replica replica = testcoll.getSlice("a").getReplicas().iterator().next();

  removeAndWaitForLastReplicaGone(collectionName, replica, "a");
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
/** Returns all replicas per node. */
private Map<String, List<ReplicaInfo>> getNodeReplicas() {
    final ClusterState clusterState = getClusterState();
    final Map<String, List<ReplicaInfo>> result = Maps.newHashMap();
    for (final DocCollection collection : clusterState.getCollectionsMap().values()) {
        for (final Slice slice : collection.getSlices()) {
            for (final Replica replica : slice.getReplicas()) {
                List<ReplicaInfo> nodeReplicas = result.get(replica.getNodeName());
                if (nodeReplicas == null) {
                    nodeReplicas = Lists.newArrayList();
                    result.put(replica.getNodeName(), nodeReplicas);
                }
                nodeReplicas.add(new ReplicaInfo(replica, collection.getName(), slice.getName()));
            }
        }
    }
    return result;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
@Test
public void testDefaultSliceState() throws Exception {
  Map<String, DocCollection> collectionStates = new HashMap<String, DocCollection>();
  Set<String> liveNodes = new HashSet<String>();
  liveNodes.add("node1");

  Map<String, Slice> slices = new HashMap<String, Slice>();
  Map<String, Replica> sliceToProps = new HashMap<String, Replica>();
  Map<String, Object> props = new HashMap<String, Object>();

  Replica replica = new Replica("node1", props);
  sliceToProps.put("node1", replica);
  Slice slice = new Slice("shard1", sliceToProps, null);
  assertEquals("Default state not set to active", Slice.ACTIVE, slice.getState());
  slices.put("shard1", slice);
  collectionStates.put("collection1", new DocCollection("collection1", slices, null, DocRouter.DEFAULT));

  ClusterState clusterState = new ClusterState(liveNodes, collectionStates);
  byte[] bytes = ZkStateReader.toJSON(clusterState);
  ClusterState loadedClusterState = ClusterState.load(null, bytes, liveNodes);

  assertEquals("Default state not set to active", "active", loadedClusterState.getSlice("collection1", "shard1").getState());
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private void waitTillRecovered() throws Exception {
  for (int i = 0; i < 30; i++) { 
    Thread.sleep(3000);
    ZkStateReader zkStateReader = cloudClient.getZkStateReader();
    zkStateReader.updateClusterState(true);
    ClusterState clusterState = zkStateReader.getClusterState();
    DocCollection collection1 = clusterState.getCollection("collection1");
    Slice slice = collection1.getSlice("shard1");
    Collection<Replica> replicas = slice.getReplicas();
    boolean allActive = true;
    for (Replica replica : replicas) {
      if (!clusterState.liveNodesContain(replica.getNodeName())
          || !replica.get(ZkStateReader.STATE_PROP).equals(
              ZkStateReader.ACTIVE)) {
        allActive = false;
        break;
      }
    }
    if (allActive) {
      return;
    }
  }
  printLayout();
  fail("timeout waiting to see recovered node");
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private void checkForMissingCollection(String collectionName)
    throws Exception {
  // check for a  collection - we poll the state
  long timeoutAt = System.currentTimeMillis() + 45000;
  boolean found = true;
  while (System.currentTimeMillis() < timeoutAt) {
    getCommonCloudSolrServer().getZkStateReader().updateClusterState(true);
    ClusterState clusterState = getCommonCloudSolrServer().getZkStateReader().getClusterState();
    Map<String,DocCollection> collections = clusterState
        .getCollectionStates();
    if (!collections.containsKey(collectionName)) {
      found = false;
      break;
    }
    Thread.sleep(100);
  }
  if (found) {
    fail("Found collection that should be gone " + collectionName);
  }
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
public List<List<String>> extractShardUrls(String zkHost, String collection) {

    DocCollection docCollection = extractDocCollection(zkHost, collection);
    List<Slice> slices = getSortedSlices(docCollection.getSlices());
    List<List<String>> solrUrls = new ArrayList<List<String>>(slices.size());
    for (Slice slice : slices) {
      if (slice.getLeader() == null) {
        throw new IllegalArgumentException("Cannot find SolrCloud slice leader. " +
            "It looks like not all of your shards are registered in ZooKeeper yet");
      }
      Collection<Replica> replicas = slice.getReplicas();
      List<String> urls = new ArrayList<String>(replicas.size());
      for (Replica replica : replicas) {
        ZkCoreNodeProps props = new ZkCoreNodeProps(replica);
        urls.add(props.getCoreUrl());
      }
      solrUrls.add(urls);
    }
    return solrUrls;
  }
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private List<Node> getSubShardLeaders(DocCollection coll, String shardId, String docId, SolrInputDocument doc) {
  Collection<Slice> allSlices = coll.getSlices();
  List<Node> nodes = null;
  for (Slice aslice : allSlices) {
    if (Slice.CONSTRUCTION.equals(aslice.getState()) || Slice.RECOVERY.equals(aslice.getState()))  {
      DocRouter.Range myRange = coll.getSlice(shardId).getRange();
      if (myRange == null) myRange = new DocRouter.Range(Integer.MIN_VALUE, Integer.MAX_VALUE);
      boolean isSubset = aslice.getRange() != null && aslice.getRange().isSubsetOf(myRange);
      if (isSubset &&
          (docId == null // in case of deletes
          || (docId != null && coll.getRouter().isTargetSlice(docId, doc, req.getParams(), aslice.getName(), coll)))) {
        Replica sliceLeader = aslice.getLeader();
        // slice leader can be null because node/shard is created zk before leader election
        if (sliceLeader != null && zkController.getClusterState().liveNodesContain(sliceLeader.getNodeName()))  {
          if (nodes == null) nodes = new ArrayList<Node>();
          ZkCoreNodeProps nodeProps = new ZkCoreNodeProps(sliceLeader);
          nodes.add(new StdNode(nodeProps));
        }
      }
    }
  }
  return nodes;
}
 

import org.apache.solr.common.cloud.DocCollection; //导入依赖的package包/类
private void collectionCmd(ClusterState clusterState, ZkNodeProps message, ModifiableSolrParams params, NamedList results, String stateMatcher) {
  log.info("Executing Collection Cmd : " + params);
  String collectionName = message.getStr("name");
  
  DocCollection coll = clusterState.getCollection(collectionName);
  
  for (Map.Entry<String,Slice> entry : coll.getSlicesMap().entrySet()) {
    Slice slice = entry.getValue();
    sliceCmd(clusterState, params, stateMatcher, slice);
  }
  
  ShardResponse srsp;
  do {
    srsp = shardHandler.takeCompletedOrError();
    if (srsp != null) {
      processResponse(results, srsp);
    }
  } while (srsp != null);

}