import org.apache.solr.handler.component.ShardResponse; //导入依赖的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.ShardResponse; //导入依赖的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.ShardResponse; //导入依赖的package包/类
private void collectShardResponses(NamedList results, boolean abortOnError,
                                   String msgOnError,
                                   ShardHandler shardHandler) {
  ShardResponse srsp;
  do {
    srsp = shardHandler.takeCompletedOrError();
    if (srsp != null) {
      processResponse(results, srsp);
      Throwable exception = srsp.getException();
      if (abortOnError && exception != null)  {
        // drain pending requests
        while (srsp != null)  {
          srsp = shardHandler.takeCompletedOrError();
        }
        throw new SolrException(ErrorCode.SERVER_ERROR, msgOnError, exception);
      }
    }
  } while (srsp != null);
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
protected List<SubmitCapture> mockShardHandlerForCreateJob(
    Integer numberOfSlices, Integer numberOfReplica) {
  List<SubmitCapture> submitCaptures = new ArrayList<>();
  for (int i = 0; i < (numberOfSlices * numberOfReplica); i++) {
    SubmitCapture submitCapture = new SubmitCapture();
    shardHandlerMock.submit(capture(submitCapture.shardRequestCapture),
        capture(submitCapture.nodeUrlsWithoutProtocolPartCapture),
        capture(submitCapture.params));
    expectLastCall();
    submitCaptures.add(submitCapture);
    ShardResponse shardResponseWithoutException = new ShardResponse();
    shardResponseWithoutException.setSolrResponse(new QueryResponse());
    expect(shardHandlerMock.takeCompletedOrError()).andReturn(
        shardResponseWithoutException);
  }
  expect(shardHandlerMock.takeCompletedOrError()).andReturn(null);
  return submitCaptures;
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的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.ShardResponse; //导入依赖的package包/类
private void processResponse(NamedList results, ShardResponse srsp) {
  Throwable e = srsp.getException();
  if (e != null) {
    log.error("Error from shard: " + srsp.getShard(), e);
    
    SimpleOrderedMap failure = (SimpleOrderedMap) results.get("failure");
    if (failure == null) {
      failure = new SimpleOrderedMap();
      results.add("failure", failure);
    }

    failure.add(srsp.getNodeName(), e.getClass().getName() + ":" + e.getMessage());
    
  } else {
    
    SimpleOrderedMap success = (SimpleOrderedMap) results.get("success");
    if (success == null) {
      success = new SimpleOrderedMap();
      results.add("success", success);
    }
    
    success.add(srsp.getNodeName(), srsp.getSolrResponse().getResponse());
  }
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
protected List<SubmitCapture> mockShardHandlerForCreateJob(
    Integer numberOfSlices, Integer numberOfReplica) {
  List<SubmitCapture> submitCaptures = new ArrayList<SubmitCapture>();
  for (int i = 0; i < (numberOfSlices * numberOfReplica); i++) {
    SubmitCapture submitCapture = new SubmitCapture();
    shardHandlerMock.submit(capture(submitCapture.shardRequestCapture),
        capture(submitCapture.nodeUrlsWithoutProtocolPartCapture),
        capture(submitCapture.params));
    expectLastCall();
    submitCaptures.add(submitCapture);
    ShardResponse shardResponseWithoutException = new ShardResponse();
    shardResponseWithoutException.setSolrResponse(new QueryResponse());
    expect(shardHandlerMock.takeCompletedOrError()).andReturn(
        shardResponseWithoutException);
  }
  expect(shardHandlerMock.takeCompletedOrError()).andReturn(null);
  return submitCaptures;
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的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.ShardResponse; //导入依赖的package包/类
private void collectShardResponses(NamedList results, boolean abortOnError, String msgOnError) {
  ShardResponse srsp;
  do {
    srsp = shardHandler.takeCompletedOrError();
    if (srsp != null) {
      processResponse(results, srsp);
      Throwable exception = srsp.getException();
      if (abortOnError && exception != null)  {
        // drain pending requests
        while (srsp != null)  {
          srsp = shardHandler.takeCompletedOrError();
        }
        throw new SolrException(ErrorCode.SERVER_ERROR, msgOnError, exception);
      }
    }
  } while (srsp != null);
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的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);

}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
private boolean handleResponse(ShardResponse srsp) {
  NamedList<Object> response = srsp.getSolrResponse().getResponse();
  // TODO: why does this return null sometimes?
  if (response == null) {
    return false;
  }
  Boolean success = (Boolean) response.get("sync");
  
  if (success == null) {
    success = false;
  }
  
  return success;
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
private void processResponses(NamedList results, ShardHandler shardHandler) {
  ShardResponse srsp;
  do {
    srsp = shardHandler.takeCompletedOrError();
    if (srsp != null) {
      processResponse(results, srsp);
    }
  } while (srsp != null);
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
private void processResponse(NamedList results, ShardResponse srsp) {
  Throwable e = srsp.getException();
  String nodeName = srsp.getNodeName();
  SolrResponse solrResponse = srsp.getSolrResponse();
  String shard = srsp.getShard();

  processResponse(results, e, nodeName, solrResponse, shard);
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的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.ShardResponse; //导入依赖的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.ShardResponse; //导入依赖的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.ShardResponse; //导入依赖的package包/类
private boolean handleResponse(ShardResponse srsp) {
  ShardRequest sreq = srsp.getShardRequest();

  if (srsp.getException() != null) {

    // TODO: look at this more thoroughly - we don't want
    // to fail on connection exceptions, but it may make sense
    // to determine this based on the number of fails
    //
    // If the replica went down between asking for versions and asking for specific updates, that
    // shouldn't be treated as success since we counted on getting those updates back (and avoided
    // redundantly asking other replicas for them).
    if (cantReachIsSuccess && sreq.purpose == 1 && srsp.getException() instanceof SolrServerException) {
      Throwable solrException = ((SolrServerException) srsp.getException())
          .getRootCause();
      boolean connectTimeoutExceptionInChain = connectTimeoutExceptionInChain(srsp.getException());
      if (connectTimeoutExceptionInChain || solrException instanceof ConnectException || solrException instanceof ConnectTimeoutException
          || solrException instanceof NoHttpResponseException || solrException instanceof SocketException) {
        log.warn(msg() + " couldn't connect to " + srsp.getShardAddress() + ", counting as success");

        return true;
      }
    }
    
    if (cantReachIsSuccess && sreq.purpose == 1 && srsp.getException() instanceof SolrException && ((SolrException) srsp.getException()).code() == 503) {
      log.warn(msg() + " got a 503 from " + srsp.getShardAddress() + ", counting as success");
      return true;
    }
    
    if (cantReachIsSuccess && sreq.purpose == 1 && srsp.getException() instanceof SolrException && ((SolrException) srsp.getException()).code() == 404) {
      log.warn(msg() + " got a 404 from " + srsp.getShardAddress() + ", counting as success. " +
          "Perhaps /get is not registered?");
      return true;
    }
    
    // TODO: we should return the above information so that when we can request a recovery through zookeeper, we do
    // that for these nodes
    
    // TODO: at least log???
    // srsp.getException().printStackTrace(System.out);
   
    log.warn(msg() + " exception talking to " + srsp.getShardAddress() + ", failed", srsp.getException());
    
    return false;
  }

  if (sreq.purpose == 1) {
    return handleVersions(srsp);
  } else {
    return handleUpdates(srsp);
  }
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
private boolean handleVersions(ShardResponse srsp) {
  // we retrieved the last N updates from the replica
  List<Long> otherVersions = (List<Long>)srsp.getSolrResponse().getResponse().get("versions");
  // TODO: how to handle short lists?

  SyncShardRequest sreq = (SyncShardRequest) srsp.getShardRequest();
  sreq.reportedVersions =  otherVersions;

  log.info(msg() + " Received " + otherVersions.size() + " versions from " + sreq.shards[0] );

  if (otherVersions.size() == 0) {
    return getNoVersionsIsSuccess; 
  }
  
  boolean completeList = otherVersions.size() < nUpdates;  // do we have their complete list of updates?

  Collections.sort(otherVersions, absComparator);

  if (debug) {
    log.debug(msg() + " sorted versions from " + sreq.shards[0] + " = " + otherVersions);
  }
  
  long otherHigh = percentile(otherVersions, .2f);
  long otherLow = percentile(otherVersions, .8f);

  if (ourHighThreshold < otherLow) {
    // Small overlap between version windows and ours is older
    // This means that we might miss updates if we attempted to use this method.
    // Since there exists just one replica that is so much newer, we must
    // fail the sync.
    log.info(msg() + " Our versions are too old. ourHighThreshold="+ourHighThreshold + " otherLowThreshold="+otherLow);
    return false;
  }

  if (ourLowThreshold > otherHigh) {
    // Small overlap between windows and ours is newer.
    // Using this list to sync would result in requesting/replaying results we don't need
    // and possibly bringing deleted docs back to life.
    log.info(msg() + " Our versions are newer. ourLowThreshold="+ourLowThreshold + " otherHigh="+otherHigh);
    return true;
  }
  
  List<Long> toRequest = new ArrayList<>();
  for (Long otherVersion : otherVersions) {
    // stop when the entries get old enough that reorders may lead us to see updates we don't need
    if (!completeList && Math.abs(otherVersion) < ourLowThreshold) break;

    if (ourUpdateSet.contains(otherVersion) || requestedUpdateSet.contains(otherVersion)) {
      // we either have this update, or already requested it
      // TODO: what if the shard we previously requested this from returns failure (because it goes
      // down)
      continue;
    }

    toRequest.add(otherVersion);
    requestedUpdateSet.add(otherVersion);
  }

  sreq.requestedUpdates = toRequest;
  
  if (toRequest.isEmpty()) {
    log.info(msg() + " Our versions are newer. ourLowThreshold="+ourLowThreshold + " otherHigh="+otherHigh);

    // we had (or already requested) all the updates referenced by the replica
    return true;
  }
  
  if (toRequest.size() > maxUpdates) {
    log.info(msg() + " Failing due to needing too many updates:" + maxUpdates);
    return false;
  }

  return requestUpdates(srsp, toRequest);
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
private boolean handleResponse(ShardResponse srsp) {
  ShardRequest sreq = srsp.getShardRequest();

  if (srsp.getException() != null) {

    // TODO: look at this more thoroughly - we don't want
    // to fail on connection exceptions, but it may make sense
    // to determine this based on the number of fails
    //
    // If the replica went down between asking for versions and asking for specific updates, that
    // shouldn't be treated as success since we counted on getting those updates back (and avoided
    // redundantly asking other replicas for them).
    if (cantReachIsSuccess && sreq.purpose == 1 && srsp.getException() instanceof SolrServerException) {
      Throwable solrException = ((SolrServerException) srsp.getException())
          .getRootCause();
      if (solrException instanceof ConnectException || solrException instanceof ConnectTimeoutException
          || solrException instanceof NoHttpResponseException || solrException instanceof SocketException) {
        log.warn(msg() + " couldn't connect to " + srsp.getShardAddress() + ", counting as success");

        return true;
      }
    }
    
    if (cantReachIsSuccess && sreq.purpose == 1 && srsp.getException() instanceof SolrException && ((SolrException) srsp.getException()).code() == 503) {
      log.warn(msg() + " got a 503 from " + srsp.getShardAddress() + ", counting as success");
      return true;
    }
    
    if (cantReachIsSuccess && sreq.purpose == 1 && srsp.getException() instanceof SolrException && ((SolrException) srsp.getException()).code() == 404) {
      log.warn(msg() + " got a 404 from " + srsp.getShardAddress() + ", counting as success");
      return true;
    }
    // TODO: at least log???
    // srsp.getException().printStackTrace(System.out);
   
    log.warn(msg() + " exception talking to " + srsp.getShardAddress() + ", failed", srsp.getException());
    
    return false;
  }

  if (sreq.purpose == 1) {
    return handleVersions(srsp);
  } else {
    return handleUpdates(srsp);
  }
}
 

import org.apache.solr.handler.component.ShardResponse; //导入依赖的package包/类
private boolean handleVersions(ShardResponse srsp) {
  // we retrieved the last N updates from the replica
  List<Long> otherVersions = (List<Long>)srsp.getSolrResponse().getResponse().get("versions");
  // TODO: how to handle short lists?

  SyncShardRequest sreq = (SyncShardRequest) srsp.getShardRequest();
  sreq.reportedVersions =  otherVersions;

  log.info(msg() + " Received " + otherVersions.size() + " versions from " + sreq.shards[0] );

  if (otherVersions.size() == 0) {
    return getNoVersionsIsSuccess; 
  }
  
  boolean completeList = otherVersions.size() < nUpdates;  // do we have their complete list of updates?

  Collections.sort(otherVersions, absComparator);

  if (debug) {
    log.debug(msg() + " sorted versions from " + sreq.shards[0] + " = " + otherVersions);
  }
  
  long otherHigh = percentile(otherVersions, .2f);
  long otherLow = percentile(otherVersions, .8f);

  if (ourHighThreshold < otherLow) {
    // Small overlap between version windows and ours is older
    // This means that we might miss updates if we attempted to use this method.
    // Since there exists just one replica that is so much newer, we must
    // fail the sync.
    log.info(msg() + " Our versions are too old. ourHighThreshold="+ourHighThreshold + " otherLowThreshold="+otherLow);
    return false;
  }

  if (ourLowThreshold > otherHigh) {
    // Small overlap between windows and ours is newer.
    // Using this list to sync would result in requesting/replaying results we don't need
    // and possibly bringing deleted docs back to life.
    log.info(msg() + " Our versions are newer. ourLowThreshold="+ourLowThreshold + " otherHigh="+otherHigh);
    return true;
  }
  
  List<Long> toRequest = new ArrayList<Long>();
  for (Long otherVersion : otherVersions) {
    // stop when the entries get old enough that reorders may lead us to see updates we don't need
    if (!completeList && Math.abs(otherVersion) < ourLowThreshold) break;

    if (ourUpdateSet.contains(otherVersion) || requestedUpdateSet.contains(otherVersion)) {
      // we either have this update, or already requested it
      // TODO: what if the shard we previously requested this from returns failure (because it goes
      // down)
      continue;
    }

    toRequest.add(otherVersion);
    requestedUpdateSet.add(otherVersion);
  }

  sreq.requestedUpdates = toRequest;
  
  if (toRequest.isEmpty()) {
    log.info(msg() + " Our versions are newer. ourLowThreshold="+ourLowThreshold + " otherHigh="+otherHigh);

    // we had (or already requested) all the updates referenced by the replica
    return true;
  }
  
  if (toRequest.size() > maxUpdates) {
    log.info(msg() + " Failing due to needing too many updates:" + maxUpdates);
    return false;
  }

  return requestUpdates(srsp, toRequest);
}