import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private boolean requestUpdates(ShardResponse srsp, List<Long> toRequest) {
  String replica = srsp.getShardRequest().shards[0];

  log.info(msg() + "Requesting updates from " + replica + "n=" + toRequest.size() + " versions=" + toRequest);

  // reuse our original request object
  ShardRequest sreq = srsp.getShardRequest();

  sreq.purpose = 0;
  sreq.params = new ModifiableSolrParams();
  sreq.params.set("qt", "/get");
  sreq.params.set("distrib", false);
  sreq.params.set("getUpdates", StrUtils.join(toRequest, ','));
  sreq.params.set("onlyIfActive", onlyIfActive);
  sreq.responses.clear();  // needs to be zeroed for correct correlation to occur

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

  return true;
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public void process(ResponseBuilder rb, ShardRequest shardRequest) {
  boolean returnScores = (rb.getFieldFlags() & SolrIndexSearcher.GET_SCORES) != 0;
  ShardResponse srsp = shardRequest.responses.get(0);
  SolrDocumentList docs = (SolrDocumentList)srsp.getSolrResponse().getResponse().get("response");
  String uniqueIdFieldName = rb.req.getSchema().getUniqueKeyField().getName();

  for (SolrDocument doc : docs) {
    Object id = doc.getFieldValue(uniqueIdFieldName).toString();
    ShardDoc shardDoc = rb.resultIds.get(id);
    FieldDoc fieldDoc = (FieldDoc) shardDoc;
    if (shardDoc != null) {
      if (returnScores && !Float.isNaN(fieldDoc.score)) {
          doc.setField("score", fieldDoc.score);
      }
      rb.retrievedDocuments.put(id, doc);
    }
  }
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private void invokeOverseerOp(String electionNode, String op) {
  ModifiableSolrParams params = new ModifiableSolrParams();
  ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
  params.set(CoreAdminParams.ACTION, CoreAdminAction.OVERSEEROP.toString());
  params.set("op", op);
  params.set("qt", adminPath);
  params.set("electionNode", electionNode);
  ShardRequest sreq = new ShardRequest();
  sreq.purpose = 1;
  String replica = zkStateReader.getBaseUrlForNodeName(LeaderElector.getNodeName(electionNode));
  sreq.shards = new String[]{replica};
  sreq.actualShards = sreq.shards;
  sreq.params = params;
  shardHandler.submit(sreq, replica, sreq.params);
  shardHandler.takeCompletedOrError();
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
/** Dispatch shard request in <code>STAGE_EXECUTE_QUERY</code> stage */
@Override
public int distributedProcess(ResponseBuilder rb) {
  SolrParams params = rb.req.getParams();
  LOG.debug("SuggestComponent distributedProcess with : " + params);
  if (rb.stage < ResponseBuilder.STAGE_EXECUTE_QUERY) 
    return ResponseBuilder.STAGE_EXECUTE_QUERY;
  if (rb.stage == ResponseBuilder.STAGE_EXECUTE_QUERY) {
    ShardRequest sreq = new ShardRequest();
    sreq.purpose = ShardRequest.PURPOSE_GET_TOP_IDS;
    sreq.params = new ModifiableSolrParams(rb.req.getParams());
    sreq.params.remove(ShardParams.SHARDS);
    rb.addRequest(this, sreq);
    return ResponseBuilder.STAGE_GET_FIELDS;
  }

  return ResponseBuilder.STAGE_DONE;
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private boolean requestUpdates(ShardResponse srsp, List<Long> toRequest) {
  String replica = srsp.getShardRequest().shards[0];

  log.info(msg() + "Requesting updates from " + replica + "n=" + toRequest.size() + " versions=" + toRequest);

  // reuse our original request object
  ShardRequest sreq = srsp.getShardRequest();

  sreq.purpose = 0;
  sreq.params = new ModifiableSolrParams();
  sreq.params.set("qt","/get");
  sreq.params.set("distrib",false);
  sreq.params.set("getUpdates", StrUtils.join(toRequest, ','));
  sreq.responses.clear();  // needs to be zeroed for correct correlation to occur

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

  return true;
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
@Override
public void modifyRequest(ResponseBuilder rb, SearchComponent who, ShardRequest sreq) {
	System.out.println("===== MODIFY REQUEST =====");
	System.out.println("who=" + who);
	System.out.println("purpose=" + sreq.purpose);
	if ((sreq.purpose & ShardRequest.PURPOSE_GET_FIELDS) > 0) {
		if (fieldListIncludes(rb, DJOIN_FIELD)) {
			Set<String> fl = new HashSet<>(getFieldList(sreq.params));
			fl.add(SHARD_FIELD);
			sreq.params.set(CommonParams.FL, String.join(",", fl));
		}
		
		// enable faceting on shards to get join ids
		sreq.params.set("facet", true);
		sreq.params.set("facet.field", joinField);
	}
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
public void handleResponses(ResponseBuilder rb, ShardRequest req) {
	System.out.println("===== HANDLE RESPONSES =====");
	System.out.println("purpose=" + req.purpose);
	System.out.println("Shards: " + (req.shards != null ? String.join(" ", req.shards) : "(null)"));
	if ((req.purpose & ShardRequest.PURPOSE_GET_FIELDS) > 0) {
		Map<String, Long> numFounds = (Map<String, Long>)rb.req.getContext().get(COMPONENT_NAME + "numFounds");
		Set<Object> joinIds = (Set<Object>)rb.req.getContext().get(COMPONENT_NAME + "joinIds");
		for (ShardResponse rsp : req.responses) {
			NamedList response = rsp.getSolrResponse().getResponse();
			SolrDocumentList results = (SolrDocumentList)response.get("response");
			numFounds.put(rsp.getShard(), results.getNumFound());
			NamedList counts = (NamedList)response.get("facet_counts");
			if (counts != null) {
				NamedList fields = (NamedList)counts.get("facet_fields");
				NamedList values = (NamedList)fields.get(joinField);
				for (int i = 0; i < values.size(); ++i) {
					joinIds.add(values.getName(i));
				}
			}
		}
	}
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
@Override
public void modifyRequest(ResponseBuilder rb, SearchComponent who, ShardRequest sreq) {
  if (!checkComponentShouldProcess(rb)) {
    return;
  }

  sreq.params.set(getComponentName(), "false");
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
@Override
public void modifyRequest(ResponseBuilder rb, SearchComponent who, ShardRequest sreq) {
  SolrParams params = rb.req.getParams();
  if (!params.getBool(COMPONENT_NAME, false) || !params.getBool(ClusteringParams.USE_SEARCH_RESULTS, false)) {
    return;
  }
  sreq.params.remove(COMPONENT_NAME);
  if( ( sreq.purpose & ShardRequest.PURPOSE_GET_FIELDS ) != 0 ){
    String fl = sreq.params.get(CommonParams.FL,"*");
    // if fl=* then we don't need check
    if( fl.indexOf( '*' ) >= 0 ) return;
    Set<String> fields = getSearchClusteringEngine(rb).getFieldsToLoad(rb.req);
    if( fields == null || fields.size() == 0 ) return;
    StringBuilder sb = new StringBuilder();
    String[] flparams = fl.split( "[,\\s]+" );
    Set<String> flParamSet = new HashSet<>(flparams.length);
    for( String flparam : flparams ){
      // no need trim() because of split() by \s+
      flParamSet.add(flparam);
    }
    for( String aFieldToLoad : fields ){
      if( !flParamSet.contains( aFieldToLoad ) ){
        sb.append( ',' ).append( aFieldToLoad );
      }
    }
    if( sb.length() > 0 ){
      sreq.params.set( CommonParams.FL, fl + sb.toString() );
    }
  }
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public ShardRequest[] constructRequest(ResponseBuilder rb) {
  // If we have a group.query we need to query all shards... Or we move this to the group first phase queries
  boolean containsGroupByQuery = rb.getGroupingSpec().getQueries().length > 0;
  // TODO: If groups.truncate=true we only have to query the specific shards even faceting and statistics are enabled
  if ((rb.getQueryCommand().getFlags() & SolrIndexSearcher.GET_DOCSET) != 0 || containsGroupByQuery) {
    // In case we need more results such as faceting and statistics we have to query all shards
    return createRequestForAllShards(rb);
  } else {
    // In case we only need top groups we only have to query the shards that contain these groups.
    return createRequestForSpecificShards(rb);
  }
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private ShardRequest[] createRequestForSpecificShards(ResponseBuilder rb) {
  // Determine all unique shards to query for TopGroups
  Set<String> uniqueShards = new HashSet<>();
  for (String command : rb.searchGroupToShards.keySet()) {
    Map<SearchGroup<BytesRef>, Set<String>> groupsToShard = rb.searchGroupToShards.get(command);
    for (Set<String> shards : groupsToShard.values()) {
      uniqueShards.addAll(shards);
    }
  }

  return createRequest(rb, uniqueShards.toArray(new String[uniqueShards.size()]));
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private void sendShardRequest(String nodeName, ModifiableSolrParams params, ShardHandler shardHandler) {
  ShardRequest sreq = new ShardRequest();
  params.set("qt", adminPath);
  sreq.purpose = 1;
  String replica = zkStateReader.getBaseUrlForNodeName(nodeName);
  sreq.shards = new String[]{replica};
  sreq.actualShards = sreq.shards;
  sreq.params = params;

  shardHandler.submit(sreq, replica, sreq.params);
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private void sliceCmd(ClusterState clusterState, ModifiableSolrParams params, String stateMatcher,
                      Slice slice, ShardHandler shardHandler) {
  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)) && (stateMatcher != null ? node.getStr(ZkStateReader.STATE_PROP).equals(stateMatcher) : true)) {
      // For thread safety, only simple clone the ModifiableSolrParams
      ModifiableSolrParams cloneParams = new ModifiableSolrParams();
      cloneParams.add(params);
      cloneParams.set(CoreAdminParams.CORE,
          node.getStr(ZkStateReader.CORE_NAME_PROP));

      String replica = node.getStr(ZkStateReader.BASE_URL_PROP);
      ShardRequest sreq = new ShardRequest();
      sreq.nodeName = node.getStr(ZkStateReader.NODE_NAME_PROP);
      // yes, they must use same admin handler path everywhere...
      cloneParams.set("qt", adminPath);
      sreq.purpose = 1;
      sreq.shards = new String[] {replica};
      sreq.actualShards = sreq.shards;
      sreq.params = cloneParams;
      log.info("Collection Admin sending CoreAdmin cmd to " + replica
          + " params:" + sreq.params);
      shardHandler.submit(sreq, replica, sreq.params);
    }
  }
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
public void merge(ResponseBuilder rb, ShardRequest shardRequest) {
  int count = 0;
  NamedList merged = new NamedList();

  for(ShardResponse shardResponse : shardRequest.responses) {
    NamedList response = shardResponse.getSolrResponse().getResponse();
    NamedList analytics = (NamedList)response.get("analytics");
    Integer c = (Integer)analytics.get("mycount");
    count += c.intValue();
  }

  merged.add("mycount", count);
  rb.rsp.add("analytics", merged);
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
@Override
public ShardHandler getShardHandler() {
  return new ShardHandler() {
    @Override
    public void checkDistributed(ResponseBuilder rb) {}

    @Override
    public void submit(ShardRequest sreq, String shard,
        ModifiableSolrParams params) {}

    @Override
    public ShardResponse takeCompletedIncludingErrors() {
      return null;
    }

    @Override
    public ShardResponse takeCompletedOrError() {
      return null;
    }

    @Override
    public void cancelAll() {}

    @Override
    public ShardHandlerFactory getShardHandlerFactory() {
      return MockShardHandlerFactory.this;
    }
  };
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
/** 
 * Used in Distributed Search, merges the suggestion results from every shard
 * */
@Override
public void finishStage(ResponseBuilder rb) {
  SolrParams params = rb.req.getParams();
  LOG.debug("SuggestComponent finishStage with : " + params);
  if (!params.getBool(COMPONENT_NAME, false) || rb.stage != ResponseBuilder.STAGE_GET_FIELDS)
    return;
  int count = params.getInt(SUGGEST_COUNT, 1);
  
  List<SuggesterResult> suggesterResults = new ArrayList<>();
  
  // Collect Shard responses
  for (ShardRequest sreq : rb.finished) {
    for (ShardResponse srsp : sreq.responses) {
      NamedList<Object> resp;
      if((resp = srsp.getSolrResponse().getResponse()) != null) {
        @SuppressWarnings("unchecked")
        Map<String, SimpleOrderedMap<NamedList<Object>>> namedList = 
            (Map<String, SimpleOrderedMap<NamedList<Object>>>) resp.get(SuggesterResultLabels.SUGGEST);
        LOG.debug(srsp.getShard() + " : " + namedList);
        suggesterResults.add(toSuggesterResult(namedList));
      }
    }
  }
  
  // Merge Shard responses
  SuggesterResult suggesterResult = merge(suggesterResults, count);
  Map<String, SimpleOrderedMap<NamedList<Object>>> namedListResults = 
      new HashMap<>();
  toNamedList(suggesterResult, namedListResults);
  
  rb.rsp.add(SuggesterResultLabels.SUGGEST, namedListResults);
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
@Override
public void modifyRequest(ResponseBuilder rb, SearchComponent who, ShardRequest sreq) {
  SolrParams params = rb.req.getParams();
  if (!params.getBool(COMPONENT_NAME, false) || !params.getBool(ClusteringParams.USE_SEARCH_RESULTS, false)) {
    return;
  }
  sreq.params.remove(COMPONENT_NAME);
  if( ( sreq.purpose & ShardRequest.PURPOSE_GET_FIELDS ) != 0 ){
    String fl = sreq.params.get(CommonParams.FL,"*");
    // if fl=* then we don't need check
    if( fl.indexOf( '*' ) >= 0 ) return;
    Set<String> fields = getSearchClusteringEngine(rb).getFieldsToLoad(rb.req);
    if( fields == null || fields.size() == 0 ) return;
    StringBuilder sb = new StringBuilder();
    String[] flparams = fl.split( "[,\\s]+" );
    Set<String> flParamSet = new HashSet<String>(flparams.length);
    for( String flparam : flparams ){
      // no need trim() because of split() by \s+
      flParamSet.add(flparam);
    }
    for( String aFieldToLoad : fields ){
      if( !flParamSet.contains( aFieldToLoad ) ){
        sb.append( ',' ).append( aFieldToLoad );
      }
    }
    if( sb.length() > 0 ){
      sreq.params.set( CommonParams.FL, fl + sb.toString() );
    }
  }
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private ShardRequest[] createRequestForSpecificShards(ResponseBuilder rb) {
  // Determine all unique shards to query for TopGroups
  Set<String> uniqueShards = new HashSet<String>();
  for (String command : rb.searchGroupToShards.keySet()) {
    Map<SearchGroup<BytesRef>, Set<String>> groupsToShard = rb.searchGroupToShards.get(command);
    for (Set<String> shards : groupsToShard.values()) {
      uniqueShards.addAll(shards);
    }
  }

  return createRequest(rb, uniqueShards.toArray(new String[uniqueShards.size()]));
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
@Override
public void modifyRequest(ResponseBuilder rb, SearchComponent who, ShardRequest sreq) {
  // do the filterQParser stuff first
  super.modifyRequest(rb, who, sreq);
  
  if (! doMerge(rb)) {
    return;
  }
  
  ReturnFields rf = rb.rsp.getReturnFields();
  if (rf.wantsAllFields()) {
    // we already have what we need since we ask for everything...
    return;
  }

  IndexSchema schema = rb.req.getCore().getLatestSchema();
  for (SchemaField field : schema.getFields().values()) {
    if (! rf.wantsField(field.getName())) {
      continue;
    }
    for (String source : schema.getCopySources(field.getName())) {
      if (rf.wantsField(source)) {
        continue;
      }
      sreq.params.add(CommonParams.FL, source);
    }
  }
}
 

import org.apache.solr.handler.component.ShardRequest; //导入依赖的package包/类
private void invokeOverseerOp(String nodeName, String op) {
  ModifiableSolrParams params = new ModifiableSolrParams();
  params.set(CoreAdminParams.ACTION, CoreAdminAction.OVERSEEROP.toString());
  params.set("op", op);
  params.set("qt", adminPath);
  ShardRequest sreq = new ShardRequest();
  sreq.purpose = 1;
  String replica = zkStateReader.getBaseUrlForNodeName(nodeName);
  sreq.shards = new String[]{replica};
  sreq.actualShards = sreq.shards;
  sreq.params = params;
  shardHandler.submit(sreq, replica, sreq.params);
}