Solr4.8.0源码分析(21)之SolrCloud的Recovery策略(二)

 1     // Fire off the requests before getting our own recent updates (for better concurrency)
 2     // This also allows us to avoid getting updates we don‘t need... if we got our updates and then got their updates, they would
 3     // have newer stuff that we also had (assuming updates are going on and are being forwarded).
 4     for (String replica : replicas) {
 5       requestVersions(replica);
 6     }
 7 
 8   private void requestVersions(String replica) {
 9     SyncShardRequest sreq = new SyncShardRequest();
10     sreq.purpose = 1;
11     sreq.shards = new String[]{replica};
12     sreq.actualShards = sreq.shards;
13     sreq.params = new ModifiableSolrParams();
14     sreq.params.set("qt","/get");
15     sreq.params.set("distrib",false);
16     sreq.params.set("getVersions",nUpdates);
17     shardHandler.submit(sreq, replica, sreq.params);
18   }

1     recentUpdates = ulog.getRecentUpdates();
2     try {
3       ourUpdates = recentUpdates.getVersions(nUpdates);
4     } finally {
5       recentUpdates.close();
6     }
7 
8     Collections.sort(ourUpdates, absComparator);

1       // now make sure that the starting updates overlap our updates
2       // there shouldn‘t be reorders, so any overlap will do.
3 
4       long smallestNewUpdate = Math.abs(ourUpdates.get(ourUpdates.size()-1));
5 
6       if (Math.abs(startingVersions.get(0)) < smallestNewUpdate) {
7         log.warn(msg() + "too many updates received since start - startingUpdates no longer overlaps with our currentUpdates");
8         return false;
9       }

1       // let‘s merge the lists
2       List<Long> newList = new ArrayList<>(ourUpdates);
3       for (Long ver : startingVersions) {
4         if (Math.abs(ver) < smallestNewUpdate) {
5           newList.add(ver);
6         }
7       }
8 
9       ourUpdates = newList;

 1     long otherHigh = percentile(otherVersions, .2f);
 2     long otherLow = percentile(otherVersions, .8f);
 3 
 4     if (ourHighThreshold < otherLow) {
 5       // Small overlap between version windows and ours is older
 6       // This means that we might miss updates if we attempted to use this method.
 7       // Since there exists just one replica that is so much newer, we must
 8       // fail the sync.
 9       log.info(msg() + " Our versions are too old. ourHighThreshold="+ourHighThreshold + " otherLowThreshold="+otherLow);
10       return false;
11     }

1     if (ourLowThreshold > otherHigh) {
2       // Small overlap between windows and ours is newer.
3       // Using this list to sync would result in requesting/replaying results we don‘t need
4       // and possibly bringing deleted docs back to life.
5       log.info(msg() + " Our versions are newer. ourLowThreshold="+ourLowThreshold + " otherHigh="+otherHigh);
6       return true;
7     }

 1     for (Long otherVersion : otherVersions) {
 2       // stop when the entries get old enough that reorders may lead us to see updates we don‘t need
 3       if (!completeList && Math.abs(otherVersion) < ourLowThreshold) break;
 4 
 5       if (ourUpdateSet.contains(otherVersion) || requestedUpdateSet.contains(otherVersion)) {
 6         // we either have this update, or already requested it
 7         // TODO: what if the shard we previously requested this from returns failure (because it goes
 8         // down)
 9         continue;
10       }
11 
12       toRequest.add(otherVersion);
13       requestedUpdateSet.add(otherVersion);
14     }

 1   private boolean requestUpdates(ShardResponse srsp, List<Long> toRequest) {
 2     String replica = srsp.getShardRequest().shards[0];
 3 
 4     log.info(msg() + "Requesting updates from " + replica + "n=" + toRequest.size() + " versions=" + toRequest);
 5 
 6     // reuse our original request object
 7     ShardRequest sreq = srsp.getShardRequest();
 8 
 9     sreq.purpose = 0;
10     sreq.params = new ModifiableSolrParams();
11     sreq.params.set("qt", "/get");
12     sreq.params.set("distrib", false);
13     sreq.params.set("getUpdates", StrUtils.join(toRequest, ‘,‘));
14     sreq.params.set("onlyIfActive", onlyIfActive);
15     sreq.responses.clear();  // needs to be zeroed for correct correlation to occur
16 
17     shardHandler.submit(sreq, sreq.shards[0], sreq.params);
18 
19     return true;
20   }