import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
/**
 * Executes a query against each live and active replica of the specified shard 
 * and aserts that the results are identical.
 *
 * @see #queryAndCompare
 */
public QueryResponse queryAndCompareReplicas(SolrParams params, String shard) 
  throws Exception {

  ArrayList<SolrServer> shardClients = new ArrayList<>(7);

  updateMappingsFromZk(jettys, clients);
  ZkStateReader zkStateReader = cloudClient.getZkStateReader();
  List<CloudJettyRunner> solrJetties = shardToJetty.get(shard);
  assertNotNull("no jetties found for shard: " + shard, solrJetties);


  for (CloudJettyRunner cjetty : solrJetties) {
    ZkNodeProps props = cjetty.info;
    String nodeName = props.getStr(ZkStateReader.NODE_NAME_PROP);
    boolean active = props.getStr(ZkStateReader.STATE_PROP).equals(ZkStateReader.ACTIVE);
    boolean live = zkStateReader.getClusterState().liveNodesContain(nodeName);
    if (active && live) {
      shardClients.add(cjetty.client.solrClient);
    }
  }
  return queryAndCompare(params, shardClients);
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
public static String getUrlFromZk(ClusterState clusterState, String collection) {
  Map<String,Slice> slices = clusterState.getCollection(collection).getSlicesMap();

  if (slices == null) {
    throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Could not find collection:" + collection);
  }

  for (Map.Entry<String,Slice> entry : slices.entrySet()) {
    Slice slice = entry.getValue();
    Map<String,Replica> shards = slice.getReplicasMap();
    Set<Map.Entry<String,Replica>> shardEntries = shards.entrySet();
    for (Map.Entry<String,Replica> shardEntry : shardEntries) {
      final ZkNodeProps node = shardEntry.getValue();
      if (clusterState.liveNodesContain(node.getStr(ZkStateReader.NODE_NAME_PROP))) {
        return ZkCoreNodeProps.getCoreUrl(node.getStr(ZkStateReader.BASE_URL_PROP), collection); //new ZkCoreNodeProps(node).getCoreUrl();
      }
    }
  }

  throw new RuntimeException("Could not find a live node for collection:" + collection);
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private int checkIfKillIsLegal(String slice, int numActive)
    throws KeeperException, InterruptedException {
  for (CloudJettyRunner cloudJetty : shardToJetty.get(slice)) {
    
    // get latest cloud state
    zkStateReader.updateClusterState(true);
    
    Slice theShards = zkStateReader.getClusterState().getSlicesMap(collection)
        .get(slice);
    
    ZkNodeProps props = theShards.getReplicasMap().get(cloudJetty.coreNodeName);
    if (props == null) {
      throw new RuntimeException("shard name " + cloudJetty.coreNodeName + " not found in " + theShards.getReplicasMap().keySet());
    }
    
    String state = props.getStr(ZkStateReader.STATE_PROP);
    String nodeName = props.getStr(ZkStateReader.NODE_NAME_PROP);
    
    if (cloudJetty.jetty.isRunning()
        && state.equals(ZkStateReader.ACTIVE)
        && zkStateReader.getClusterState().liveNodesContain(nodeName)) {
      numActive++;
    }
  }
  return numActive;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
public boolean sync(ZkController zkController, SolrCore core, ZkNodeProps leaderProps, boolean peerSyncOnlyWithActive) {
  if (SKIP_AUTO_RECOVERY) {
    return true;
  }
  boolean success;
  SolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams());
  SolrQueryResponse rsp = new SolrQueryResponse();
  SolrRequestInfo.setRequestInfo(new SolrRequestInfo(req, rsp));
  try {
    if (isClosed) {
      log.warn("Closed, skipping sync up.");
      return false;
    }
    log.info("Sync replicas to " + ZkCoreNodeProps.getCoreUrl(leaderProps));
    
    if (core.getUpdateHandler().getUpdateLog() == null) {
      log.error("No UpdateLog found - cannot sync");
      return false;
    }

    success = syncReplicas(zkController, core, leaderProps, peerSyncOnlyWithActive);
  } finally {
    SolrRequestInfo.clearRequestInfo();
  }
  return success;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private boolean syncWithReplicas(ZkController zkController, SolrCore core,
    ZkNodeProps props, String collection, String shardId, boolean peerSyncOnlyWithActive) {
  List<ZkCoreNodeProps> nodes = zkController.getZkStateReader()
      .getReplicaProps(collection, shardId,core.getCoreDescriptor().getCloudDescriptor().getCoreNodeName());
  
  if (nodes == null) {
    // I have no replicas
    return true;
  }
  
  List<String> syncWith = new ArrayList<>();
  for (ZkCoreNodeProps node : nodes) {
    syncWith.add(node.getCoreUrl());
  }
  
  // if we can't reach a replica for sync, we still consider the overall sync a success
  // TODO: as an assurance, we should still try and tell the sync nodes that we couldn't reach
  // to recover once more?
  PeerSync peerSync = new PeerSync(core, syncWith, core.getUpdateHandler().getUpdateLog().numRecordsToKeep, true, true, peerSyncOnlyWithActive);
  return peerSync.sync();
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
public void checkOverseerDesignate() {
  try {
    byte[] data = zkClient.getData(ZkStateReader.ROLES, null, new Stat(), true);
    if(data ==null) return;
    Map roles = (Map) ZkStateReader.fromJSON(data);
    if(roles ==null) return;
    List nodeList= (List) roles.get("overseer");
    if(nodeList == null) return;
    if(nodeList.contains(getNodeName())){
      ZkNodeProps props = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.ADDROLE.toString().toLowerCase(Locale.ROOT),
          "node", getNodeName(),
          "role", "overseer");
      log.info("Going to add role {} ",props);
      getOverseerCollectionQueue().offer(ZkStateReader.toJSON(props));
    }
  } catch (NoNodeException nne){
    return;
  } catch (Exception e) {
    log.warn("could not readd the overseer designate ",e);
  }
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private void handleProp(ZkNodeProps message)  {
  String name = message.getStr("name");
  String val = message.getStr("val");
  Map m =  reader.getClusterProps();
  if(val ==null) m.remove(name);
  else m.put(name,val);

  try {
    if(reader.getZkClient().exists(ZkStateReader.CLUSTER_PROPS,true))
      reader.getZkClient().setData(ZkStateReader.CLUSTER_PROPS,ZkStateReader.toJSON(m),true);
    else
      reader.getZkClient().create(ZkStateReader.CLUSTER_PROPS, ZkStateReader.toJSON(m),CreateMode.PERSISTENT, true);
    clusterProps = reader.getClusterProps();
  } catch (Exception e) {
    log.error("Unable to set cluster property", e);

  }
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private ClusterState createReplica(ClusterState clusterState, ZkNodeProps message) {
  log.info("createReplica() {} ", message);
  String coll = message.getStr(ZkStateReader.COLLECTION_PROP);
  if (!checkCollectionKeyExistence(message)) return clusterState;
  String slice = message.getStr(ZkStateReader.SHARD_ID_PROP);
  Slice sl = clusterState.getSlice(coll, slice);
  if(sl == null){
    log.error("Invalid Collection/Slice {}/{} ",coll,slice);
    return clusterState;
  }

  String coreNodeName = Assign.assignNode(coll, clusterState);
  Replica replica = new Replica(coreNodeName,
      makeMap(
      ZkStateReader.CORE_NAME_PROP, message.getStr(ZkStateReader.CORE_NAME_PROP),
      ZkStateReader.BASE_URL_PROP,message.getStr(ZkStateReader.BASE_URL_PROP),
      ZkStateReader.STATE_PROP,message.getStr(ZkStateReader.STATE_PROP)));
  sl.getReplicasMap().put(coreNodeName, replica);
  return clusterState;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private ClusterState buildCollection(ClusterState clusterState, ZkNodeProps message) {
  String collection = message.getStr("name");
  log.info("building a new collection: " + collection);
  if(clusterState.hasCollection(collection) ){
    log.warn("Collection {} already exists. exit" ,collection);
    return clusterState;
  }

  ArrayList<String> shardNames = new ArrayList<>();

  if(ImplicitDocRouter.NAME.equals( message.getStr("router.name",DocRouter.DEFAULT_NAME))){
    getShardNames(shardNames,message.getStr("shards",DocRouter.DEFAULT_NAME));
  } else {
    int numShards = message.getInt(ZkStateReader.NUM_SHARDS_PROP, -1);
    if(numShards<1) throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,"numShards is a required parameter for 'compositeId' router");
    getShardNames(numShards, shardNames);
  }

  return createCollection(clusterState,collection,shardNames,message);
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private ClusterState updateShardState(ClusterState clusterState, ZkNodeProps message) {
  String collection = message.getStr(ZkStateReader.COLLECTION_PROP);
  if (!checkCollectionKeyExistence(message)) return clusterState;
  log.info("Update shard state invoked for collection: " + collection + " with message: " + message);
  for (String key : message.keySet()) {
    if (ZkStateReader.COLLECTION_PROP.equals(key)) continue;
    if (QUEUE_OPERATION.equals(key)) continue;

    Slice slice = clusterState.getSlice(collection, key);
    if (slice == null)  {
      throw new RuntimeException("Overseer.updateShardState unknown collection: " + collection + " slice: " + key);
    }
    log.info("Update shard state " + key + " to " + message.getStr(key));
    Map<String, Object> props = slice.shallowCopy();
    if (Slice.RECOVERY.equals(props.get(Slice.STATE)) && Slice.ACTIVE.equals(message.getStr(key))) {
      props.remove(Slice.PARENT);
    }
    props.put(Slice.STATE, message.getStr(key));
    Slice newSlice = new Slice(slice.getName(), slice.getReplicasCopy(), props);
    clusterState = updateSlice(clusterState, collection, newSlice);
  }

  return clusterState;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private ClusterState removeRoutingRule(ClusterState clusterState, ZkNodeProps message) {
  String collection = message.getStr(ZkStateReader.COLLECTION_PROP);
  if (!checkCollectionKeyExistence(message)) return clusterState;
  String shard = message.getStr(ZkStateReader.SHARD_ID_PROP);
  String routeKeyStr = message.getStr("routeKey");

  log.info("Overseer.removeRoutingRule invoked for collection: " + collection
      + " shard: " + shard + " routeKey: " + routeKeyStr);

  Slice slice = clusterState.getSlice(collection, shard);
  if (slice == null)  {
    log.warn("Unknown collection: " + collection + " shard: " + shard);
    return clusterState;
  }
  Map<String, RoutingRule> routingRules = slice.getRoutingRules();
  if (routingRules != null) {
    routingRules.remove(routeKeyStr); // no rules left
    Map<String, Object> props = slice.shallowCopy();
    props.put("routingRules", routingRules);
    Slice newSlice = new Slice(slice.getName(), slice.getReplicasCopy(), props);
    clusterState = updateSlice(clusterState, collection, newSlice);
  }

  return clusterState;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private ClusterState updateStateNew(ClusterState clusterState, ZkNodeProps message) {
  String collection = message.getStr(ZkStateReader.COLLECTION_PROP);
  if (!checkCollectionKeyExistence(message)) return clusterState;
  String sliceName = message.getStr(ZkStateReader.SHARD_ID_PROP);

  if(collection==null || sliceName == null){
    log.error("Invalid collection and slice {}", message);
    return clusterState;
  }
  Slice slice = clusterState.getSlice(collection, sliceName);
  if(slice == null){
    log.error("No such slice exists {}", message);
    return clusterState;
  }

  return updateState(clusterState, message);
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private String getAssignedCoreNodeName(ClusterState state, ZkNodeProps message) {
  Collection<Slice> slices = state.getSlices(message.getStr(ZkStateReader.COLLECTION_PROP));
  if (slices != null) {
    for (Slice slice : slices) {
      for (Replica replica : slice.getReplicas()) {
        String nodeName = replica.getStr(ZkStateReader.NODE_NAME_PROP);
        String core = replica.getStr(ZkStateReader.CORE_NAME_PROP);
        
        String msgNodeName = message.getStr(ZkStateReader.NODE_NAME_PROP);
        String msgCore = message.getStr(ZkStateReader.CORE_NAME_PROP);
        
        if (nodeName.equals(msgNodeName) && core.equals(msgCore)) {
          return replica.getName();
        }
      }
    }
  }
  return null;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private boolean checkExclusivity(ZkNodeProps message, String id) throws KeeperException, InterruptedException {
  String collectionName = message.containsKey(COLLECTION_PROP) ?
      message.getStr(COLLECTION_PROP) : message.getStr("name");

  if(collectionName == null)
    return true;

  // CLUSTERSTATUS is always mutually exclusive
  if(CLUSTERSTATUS.isEqual(message.getStr(Overseer.QUEUE_OPERATION)))
    return true;

  if(collectionWip.contains(collectionName))
    return false;

  if(runningZKTasks.contains(id))
    return false;

  return true;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private String getConfigName(String coll, ZkNodeProps message) throws KeeperException, InterruptedException {
  String configName = message.getStr(OverseerCollectionProcessor.COLL_CONF);
  
  if(configName == null) {
    // if there is only one conf, use that
    List<String> configNames = null;
    try {
      configNames = zkStateReader.getZkClient().getChildren(ZkController.CONFIGS_ZKNODE, null, true);
      if (configNames != null && configNames.size() == 1) {
        configName = configNames.get(0);
        // no config set named, but there is only 1 - use it
        log.info("Only one config set found in zk - using it:" + configName);
      }
    } catch (KeeperException.NoNodeException e) {

    }
  }
  return configName;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private void markTaskAsRunning(QueueEvent head, String collectionName,
                               String asyncId, ZkNodeProps message)
    throws KeeperException, InterruptedException {
  synchronized (runningZKTasks) {
    runningZKTasks.add(head.getId());
  }

  synchronized (runningTasks) {
    runningTasks.add(head.getId());
  }
  if(!CLUSTERSTATUS.isEqual(message.getStr(Overseer.QUEUE_OPERATION)) && collectionName != null) {
    synchronized (collectionWip) {
      collectionWip.add(collectionName);
    }
  }

  if(asyncId != null)
    runningMap.put(asyncId, null);
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
/**
 * Returns true if the queue contains a task with the specified async id.
 */
public boolean containsTaskWithRequestId(String requestId)
    throws KeeperException, InterruptedException {

  List<String> childNames = zookeeper.getChildren(dir, null, true);
  for (String childName : childNames) {
    if (childName != null) {
      try {
        byte[] data = zookeeper.getData(dir + "/" + childName, null, null, true);
        if (data != null) {
          ZkNodeProps message = ZkNodeProps.load(data);
          if (message.containsKey(OverseerCollectionProcessor.ASYNC)) {
            LOG.debug(">>>> {}", message.get(OverseerCollectionProcessor.ASYNC));
            if(message.get(OverseerCollectionProcessor.ASYNC).equals(requestId)) return true;
          }
        }
      } catch (KeeperException.NoNodeException e) {
        // Another client removed the node first, try next
      }
    }
  }

  return false;
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
public ClientThread(ElectorSetup es, String shard, int nodeNumber, long runLeaderDelay) throws Exception {
  super("Thread-" + shard + nodeNumber);
  this.shard = shard;
  this.nodeName = shard + nodeNumber;
  this.runLeaderDelay = runLeaderDelay;

  props = new ZkNodeProps(ZkStateReader.BASE_URL_PROP, Integer.toString(nodeNumber), ZkStateReader.CORE_NAME_PROP, "");

  this.es = es;
  if (this.es == null) {
    this.es = new ElectorSetup(new OnReconnect() {
      @Override
      public void command() {
        try {
          setupOnConnect();
        } catch (Throwable t) {
        }
      }
    });
  }
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的package包/类
private String getLeaderUrl(final String collection, final String slice)
    throws KeeperException, InterruptedException {
  int iterCount = 60;
  while (iterCount-- > 0) {
    try {
      byte[] data = zkClient.getData(
          ZkStateReader.getShardLeadersPath(collection, slice), null, null,
          true);
      ZkCoreNodeProps leaderProps = new ZkCoreNodeProps(
          ZkNodeProps.load(data));
      return leaderProps.getCoreUrl();
    } catch (NoNodeException e) {
      Thread.sleep(500);
    }
  }
  zkClient.printLayoutToStdOut();
  throw new RuntimeException("Could not get leader props");
}
 

import org.apache.solr.common.cloud.ZkNodeProps; //导入依赖的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");
}