core/org.eclipse.smila.solr/lib/source/org/apache/solr/update/PeerSync.java - smila/org.eclipse.smila.core - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.solr.update;

 import static org.apache.solr.update.processor.DistributedUpdateProcessor.DistribPhase.FROMLEADER;
 import static org.apache.solr.update.processor.DistributingUpdateProcessorFactory.DISTRIB_UPDATE_PARAM;

 import java.io.IOException;
 import java.net.ConnectException;
 import java.net.SocketException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;

 import org.apache.http.NoHttpResponseException;
 import org.apache.http.client.HttpClient;
 import org.apache.http.conn.ConnectTimeoutException;
 import org.apache.lucene.util.BytesRef;
 import org.apache.solr.client.solrj.SolrServerException;
 import org.apache.solr.cloud.ZkController;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.SolrInputDocument;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.util.StrUtils;
 import org.apache.solr.core.SolrCore;
 import org.apache.solr.handler.component.HttpShardHandlerFactory;
 import org.apache.solr.handler.component.ShardHandler;
 import org.apache.solr.handler.component.ShardHandlerFactory;
 import org.apache.solr.handler.component.ShardRequest;
 import org.apache.solr.handler.component.ShardResponse;
 import org.apache.solr.request.LocalSolrQueryRequest;
 import org.apache.solr.request.SolrQueryRequest;
 import org.apache.solr.response.SolrQueryResponse;
 import org.apache.solr.update.processor.UpdateRequestProcessor;
 import org.apache.solr.update.processor.UpdateRequestProcessorChain;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /** @lucene.experimental */
 public class PeerSync  {
   public static Logger log = LoggerFactory.getLogger(PeerSync.class);
   public boolean debug = log.isDebugEnabled();

   private List<String> replicas;
   private int nUpdates;
   private int maxUpdates;  // maximum number of updates to request before failing

   private UpdateHandler uhandler;
   private UpdateLog ulog;
   private HttpShardHandlerFactory shardHandlerFactory;
   private ShardHandler shardHandler;

   private UpdateLog.RecentUpdates recentUpdates;
   private List<Long> startingVersions;

   private List<Long> ourUpdates;
   private Set<Long> ourUpdateSet;
   private Set<Long> requestedUpdateSet;
   private long ourLowThreshold;  // 20th percentile
   private long ourHighThreshold; // 80th percentile
   private final boolean cantReachIsSuccess;
   private final boolean getNoVersionsIsSuccess;
   private final HttpClient client;
   private final boolean onlyIfActive;

   // comparator that sorts by absolute value, putting highest first
   private static Comparator<Long> absComparator = new Comparator<Long>() {
     @Override
     public int compare(Long o1, Long o2) {
       long l1 = Math.abs(o1);
       long l2 = Math.abs(o2);
       if (l1 >l2) return -1;
       if (l1 < l2) return 1;
       return 0;
     }
   };

   // comparator that sorts update records by absolute value of version, putting lowest first
   private static Comparator<Object> updateRecordComparator = new Comparator<Object>() {
     @Override
     public int compare(Object o1, Object o2) {
       if (!(o1 instanceof List)) return 1;
       if (!(o2 instanceof List)) return -1;

       List lst1 = (List)o1;
       List lst2 = (List)o2;

       long l1 = Math.abs((Long)lst1.get(1));
       long l2 = Math.abs((Long)lst2.get(1));

       if (l1 >l2) return 1;
       if (l1 < l2) return -1;
       return 0;
     }
   };


   private static class SyncShardRequest extends ShardRequest {
     List<Long> reportedVersions;
     List<Long> requestedUpdates;
     Exception updateException;
   }

   public PeerSync(SolrCore core, List<String> replicas, int nUpdates) {
     this(core, replicas, nUpdates, false, true);
   }

   public PeerSync(SolrCore core, List<String> replicas, int nUpdates, boolean cantReachIsSuccess, boolean getNoVersionsIsSuccess) {
     this(core, replicas, nUpdates, cantReachIsSuccess, getNoVersionsIsSuccess, false);
   }

   public PeerSync(SolrCore core, List<String> replicas, int nUpdates, boolean cantReachIsSuccess, boolean getNoVersionsIsSuccess, boolean onlyIfActive) {
     this.replicas = replicas;
     this.nUpdates = nUpdates;
     this.maxUpdates = nUpdates;
     this.cantReachIsSuccess = cantReachIsSuccess;
     this.getNoVersionsIsSuccess = getNoVersionsIsSuccess;
     this.client = core.getCoreDescriptor().getCoreContainer().getUpdateShardHandler().getHttpClient();
     this.onlyIfActive = onlyIfActive;

     uhandler = core.getUpdateHandler();
     ulog = uhandler.getUpdateLog();
     // TODO: shutdown
     shardHandlerFactory = (HttpShardHandlerFactory) core.getCoreDescriptor().getCoreContainer().getShardHandlerFactory();
     shardHandler = shardHandlerFactory.getShardHandler(client);
   }

   /** optional list of updates we had before possibly receiving new updates */
   public void setStartingVersions(List<Long> startingVersions) {
     this.startingVersions = startingVersions;
   }

   public long percentile(List<Long> arr, float frac) {
     int elem = (int) (arr.size() * frac);
     return Math.abs(arr.get(elem));
   }

   // start of peersync related debug messages.  includes the core name for correlation.
   private String msg() {
     ZkController zkController = uhandler.core.getCoreDescriptor().getCoreContainer().getZkController();

     String myURL = "";

     if (zkController != null) {
       myURL = zkController.getBaseUrl();
     }

     // TODO: core name turns up blank in many tests - find URL if cloud enabled?
     return "PeerSync: core="+uhandler.core.getName()+ " url="+myURL +" ";
   }

   /** Returns true if peer sync was successful, meaning that this core may not be considered to have the latest updates
    *  when considering the last N updates between it and it's peers.
    *  A commit is not performed.
    */
   public boolean sync() {
     if (ulog == null) {
       return false;
     }

     log.info(msg() + "START replicas=" + replicas + " nUpdates=" + nUpdates);

     // TODO: does it ever make sense to allow sync when buffering or applying buffered?  Someone might request that we do it...
     if (!(ulog.getState() == UpdateLog.State.ACTIVE || ulog.getState()==UpdateLog.State.REPLAYING)) {
       log.error(msg() + "ERROR, update log not in ACTIVE or REPLAY state. " + ulog);
       // return false;
     }

     if (debug) {
       if (startingVersions != null) {
         log.debug(msg() + "startingVersions=" + startingVersions.size() + " " + startingVersions);
       }
     }

     // Fire off the requests before getting our own recent updates (for better concurrency)
     // This also allows us to avoid getting updates we don't need... if we got our updates and then got their updates, they would
     // have newer stuff that we also had (assuming updates are going on and are being forwarded).
     for (String replica : replicas) {
       requestVersions(replica);
     }

     recentUpdates = ulog.getRecentUpdates();
     try {
       ourUpdates = recentUpdates.getVersions(nUpdates);
     } finally {
       recentUpdates.close();
     }

     Collections.sort(ourUpdates, absComparator);

     if (startingVersions != null) {
       if (startingVersions.size() == 0) {
         log.warn("no frame of reference to tell if we've missed updates");
         return false;
       }
       Collections.sort(startingVersions, absComparator);

       ourLowThreshold = percentile(startingVersions, 0.8f);
       ourHighThreshold = percentile(startingVersions, 0.2f);

       // now make sure that the starting updates overlap our updates
       // there shouldn't be reorders, so any overlap will do.

       long smallestNewUpdate = Math.abs(ourUpdates.get(ourUpdates.size()-1));

       if (Math.abs(startingVersions.get(0)) < smallestNewUpdate) {
         log.warn(msg() + "too many updates received since start - startingUpdates no longer overlaps with our currentUpdates");
         return false;
       }

       // let's merge the lists
       List<Long> newList = new ArrayList<>(ourUpdates);
       for (Long ver : startingVersions) {
         if (Math.abs(ver) < smallestNewUpdate) {
           newList.add(ver);
         }
       }

       ourUpdates = newList;
     }  else {

       if (ourUpdates.size() > 0) {
         ourLowThreshold = percentile(ourUpdates, 0.8f);
         ourHighThreshold = percentile(ourUpdates, 0.2f);
       }  else {
         // we have no versions and hence no frame of reference to tell if we can use a peers
         // updates to bring us into sync
         log.info(msg() + "DONE.  We have no versions.  sync failed.");
         return false;
       }
     }

     ourUpdateSet = new HashSet<>(ourUpdates);
     requestedUpdateSet = new HashSet<>(ourUpdates);

     for(;;) {
       ShardResponse srsp = shardHandler.takeCompletedOrError();
       if (srsp == null) break;
       boolean success = handleResponse(srsp);
       if (!success) {
         log.info(msg() +  "DONE. sync failed");
         shardHandler.cancelAll();
         return false;
       }
     }

     log.info(msg() +  "DONE. sync succeeded");
     return true;
   }

   private void requestVersions(String replica) {
     SyncShardRequest sreq = new SyncShardRequest();
     sreq.purpose = 1;
     sreq.shards = new String[]{replica};
     sreq.actualShards = sreq.shards;
     sreq.params = new ModifiableSolrParams();
     sreq.params.set("qt","/get");
     sreq.params.set("distrib",false);
     sreq.params.set("getVersions",nUpdates);
     shardHandler.submit(sreq, replica, sreq.params);
   }

   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);
     }
   }

   // sometimes the root exception is a SocketTimeoutException, but ConnectTimeoutException
   // is in the chain
   private boolean connectTimeoutExceptionInChain(Throwable exception) {
     Throwable t = exception;
     while (true) {
       if (t instanceof ConnectTimeoutException) {
         return true;
       }
       Throwable cause = t.getCause();
       if (cause != null) {
         t = cause;
       } else {
         return false;
       }
     }
   }

   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);
   }

   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;
   }


   private boolean handleUpdates(ShardResponse srsp) {
     // we retrieved the last N updates from the replica
     List<Object> updates = (List<Object>)srsp.getSolrResponse().getResponse().get("updates");

     SyncShardRequest sreq = (SyncShardRequest) srsp.getShardRequest();
     if (updates.size() < sreq.requestedUpdates.size()) {
       log.error(msg() + " Requested " + sreq.requestedUpdates.size() + " updates from " + sreq.shards[0] + " but retrieved " + updates.size());
       return false;
     }

     ModifiableSolrParams params = new ModifiableSolrParams();
     params.set(DISTRIB_UPDATE_PARAM, FROMLEADER.toString());
     params.set("peersync",true); // debugging
     SolrQueryRequest req = new LocalSolrQueryRequest(uhandler.core, params);
     SolrQueryResponse rsp = new SolrQueryResponse();

     UpdateRequestProcessorChain processorChain = req.getCore().getUpdateProcessingChain(null);
     UpdateRequestProcessor proc = processorChain.createProcessor(req, rsp);

     Collections.sort(updates, updateRecordComparator);

     Object o = null;
     long lastVersion = 0;
     try {
       // Apply oldest updates first
       for (Object obj : updates) {
         // should currently be a List<Oper,Ver,Doc/Id>
         o = obj;
         List<Object> entry = (List<Object>)o;

         if (debug) {
           log.debug(msg() + "raw update record " + o);
         }

         int oper = (Integer)entry.get(0) & UpdateLog.OPERATION_MASK;
         long version = (Long) entry.get(1);
         if (version == lastVersion && version != 0) continue;
         lastVersion = version;

         switch (oper) {
           case UpdateLog.ADD:
           {
             // byte[] idBytes = (byte[]) entry.get(2);
             SolrInputDocument sdoc = (SolrInputDocument)entry.get(entry.size()-1);
             AddUpdateCommand cmd = new AddUpdateCommand(req);
             // cmd.setIndexedId(new BytesRef(idBytes));
             cmd.solrDoc = sdoc;
             cmd.setVersion(version);
             cmd.setFlags(UpdateCommand.PEER_SYNC | UpdateCommand.IGNORE_AUTOCOMMIT);
             if (debug) {
               log.debug(msg() + "add " + cmd);
             }
             proc.processAdd(cmd);
             break;
           }
           case UpdateLog.DELETE:
           {
             byte[] idBytes = (byte[]) entry.get(2);
             DeleteUpdateCommand cmd = new DeleteUpdateCommand(req);
             cmd.setIndexedId(new BytesRef(idBytes));
             cmd.setVersion(version);
             cmd.setFlags(UpdateCommand.PEER_SYNC | UpdateCommand.IGNORE_AUTOCOMMIT);
             if (debug) {
               log.debug(msg() + "delete " + cmd);
             }
             proc.processDelete(cmd);
             break;
           }

           case UpdateLog.DELETE_BY_QUERY:
           {
             String query = (String)entry.get(2);
             DeleteUpdateCommand cmd = new DeleteUpdateCommand(req);
             cmd.query = query;
             cmd.setVersion(version);
             cmd.setFlags(UpdateCommand.PEER_SYNC | UpdateCommand.IGNORE_AUTOCOMMIT);
             if (debug) {
               log.debug(msg() + "deleteByQuery " + cmd);
             }
             proc.processDelete(cmd);
             break;
           }

           default:
             throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,  "Unknown Operation! " + oper);
         }

       }

     }
     catch (IOException e) {
       // TODO: should this be handled separately as a problem with us?
       // I guess it probably already will by causing replication to be kicked off.
       sreq.updateException = e;
       log.error(msg() + "Error applying updates from " + sreq.shards + " ,update=" + o, e);
       return false;
     }
     catch (Exception e) {
       sreq.updateException = e;
       log.error(msg() + "Error applying updates from " + sreq.shards + " ,update=" + o, e);
       return false;
     }
     finally {
       try {
         proc.finish();
       } catch (Exception e) {
         sreq.updateException = e;
         log.error(msg() + "Error applying updates from " + sreq.shards + " ,finish()", e);
         return false;
       }
     }

     return true;
   }


   /** Requests and applies recent updates from peers */
   public static void sync(SolrCore core, List<String> replicas, int nUpdates) {
     ShardHandlerFactory shardHandlerFactory = core.getCoreDescriptor().getCoreContainer().getShardHandlerFactory();

     ShardHandler shardHandler = shardHandlerFactory.getShardHandler();

     for (String replica : replicas) {
       ShardRequest sreq = new ShardRequest();
       sreq.shards = new String[]{replica};
       sreq.params = new ModifiableSolrParams();
       sreq.params.set("qt","/get");
       sreq.params.set("distrib", false);
       sreq.params.set("getVersions",nUpdates);
       shardHandler.submit(sreq, replica, sreq.params);
     }

     for (String replica : replicas) {
       ShardResponse srsp = shardHandler.takeCompletedOrError();
     }

   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.solr.update;

	import static org.apache.solr.update.processor.DistributedUpdateProcessor.DistribPhase.FROMLEADER;
	import static org.apache.solr.update.processor.DistributingUpdateProcessorFactory.DISTRIB_UPDATE_PARAM;

	import java.io.IOException;
	import java.net.ConnectException;
	import java.net.SocketException;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.Comparator;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Set;

	import org.apache.http.NoHttpResponseException;
	import org.apache.http.client.HttpClient;
	import org.apache.http.conn.ConnectTimeoutException;
	import org.apache.lucene.util.BytesRef;
	import org.apache.solr.client.solrj.SolrServerException;
	import org.apache.solr.cloud.ZkController;
	import org.apache.solr.common.SolrException;
	import org.apache.solr.common.SolrInputDocument;
	import org.apache.solr.common.params.ModifiableSolrParams;
	import org.apache.solr.common.util.StrUtils;
	import org.apache.solr.core.SolrCore;
	import org.apache.solr.handler.component.HttpShardHandlerFactory;
	import org.apache.solr.handler.component.ShardHandler;
	import org.apache.solr.handler.component.ShardHandlerFactory;
	import org.apache.solr.handler.component.ShardRequest;
	import org.apache.solr.handler.component.ShardResponse;
	import org.apache.solr.request.LocalSolrQueryRequest;
	import org.apache.solr.request.SolrQueryRequest;
	import org.apache.solr.response.SolrQueryResponse;
	import org.apache.solr.update.processor.UpdateRequestProcessor;
	import org.apache.solr.update.processor.UpdateRequestProcessorChain;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/** @lucene.experimental */
	public class PeerSync {
	public static Logger log = LoggerFactory.getLogger(PeerSync.class);
	public boolean debug = log.isDebugEnabled();

	private List<String> replicas;
	private int nUpdates;
	private int maxUpdates; // maximum number of updates to request before failing

	private UpdateHandler uhandler;
	private UpdateLog ulog;
	private HttpShardHandlerFactory shardHandlerFactory;
	private ShardHandler shardHandler;

	private UpdateLog.RecentUpdates recentUpdates;
	private List<Long> startingVersions;

	private List<Long> ourUpdates;
	private Set<Long> ourUpdateSet;
	private Set<Long> requestedUpdateSet;
	private long ourLowThreshold; // 20th percentile
	private long ourHighThreshold; // 80th percentile
	private final boolean cantReachIsSuccess;
	private final boolean getNoVersionsIsSuccess;
	private final HttpClient client;
	private final boolean onlyIfActive;

	// comparator that sorts by absolute value, putting highest first
	private static Comparator<Long> absComparator = new Comparator<Long>() {
	@Override
	public int compare(Long o1, Long o2) {
	long l1 = Math.abs(o1);
	long l2 = Math.abs(o2);
	if (l1 >l2) return -1;
	if (l1 < l2) return 1;
	return 0;
	}
	};

	// comparator that sorts update records by absolute value of version, putting lowest first
	private static Comparator<Object> updateRecordComparator = new Comparator<Object>() {
	@Override
	public int compare(Object o1, Object o2) {
	if (!(o1 instanceof List)) return 1;
	if (!(o2 instanceof List)) return -1;

	List lst1 = (List)o1;
	List lst2 = (List)o2;

	long l1 = Math.abs((Long)lst1.get(1));
	long l2 = Math.abs((Long)lst2.get(1));

	if (l1 >l2) return 1;
	if (l1 < l2) return -1;
	return 0;
	}
	};


	private static class SyncShardRequest extends ShardRequest {
	List<Long> reportedVersions;
	List<Long> requestedUpdates;
	Exception updateException;
	}

	public PeerSync(SolrCore core, List<String> replicas, int nUpdates) {
	this(core, replicas, nUpdates, false, true);
	}

	public PeerSync(SolrCore core, List<String> replicas, int nUpdates, boolean cantReachIsSuccess, boolean getNoVersionsIsSuccess) {
	this(core, replicas, nUpdates, cantReachIsSuccess, getNoVersionsIsSuccess, false);
	}

	public PeerSync(SolrCore core, List<String> replicas, int nUpdates, boolean cantReachIsSuccess, boolean getNoVersionsIsSuccess, boolean onlyIfActive) {
	this.replicas = replicas;
	this.nUpdates = nUpdates;
	this.maxUpdates = nUpdates;
	this.cantReachIsSuccess = cantReachIsSuccess;
	this.getNoVersionsIsSuccess = getNoVersionsIsSuccess;
	this.client = core.getCoreDescriptor().getCoreContainer().getUpdateShardHandler().getHttpClient();
	this.onlyIfActive = onlyIfActive;

	uhandler = core.getUpdateHandler();
	ulog = uhandler.getUpdateLog();
	// TODO: shutdown
	shardHandlerFactory = (HttpShardHandlerFactory) core.getCoreDescriptor().getCoreContainer().getShardHandlerFactory();
	shardHandler = shardHandlerFactory.getShardHandler(client);
	}

	/** optional list of updates we had before possibly receiving new updates */
	public void setStartingVersions(List<Long> startingVersions) {
	this.startingVersions = startingVersions;
	}

	public long percentile(List<Long> arr, float frac) {
	int elem = (int) (arr.size() * frac);
	return Math.abs(arr.get(elem));
	}

	// start of peersync related debug messages. includes the core name for correlation.
	private String msg() {
	ZkController zkController = uhandler.core.getCoreDescriptor().getCoreContainer().getZkController();

	String myURL = "";

	if (zkController != null) {
	myURL = zkController.getBaseUrl();
	}

	// TODO: core name turns up blank in many tests - find URL if cloud enabled?
	return "PeerSync: core="+uhandler.core.getName()+ " url="+myURL +" ";
	}

	/** Returns true if peer sync was successful, meaning that this core may not be considered to have the latest updates
	* when considering the last N updates between it and it's peers.
	* A commit is not performed.
	*/
	public boolean sync() {
	if (ulog == null) {
	return false;
	}

	log.info(msg() + "START replicas=" + replicas + " nUpdates=" + nUpdates);

	// TODO: does it ever make sense to allow sync when buffering or applying buffered? Someone might request that we do it...
	if (!(ulog.getState() == UpdateLog.State.ACTIVE \|\| ulog.getState()==UpdateLog.State.REPLAYING)) {
	log.error(msg() + "ERROR, update log not in ACTIVE or REPLAY state. " + ulog);
	// return false;
	}

	if (debug) {
	if (startingVersions != null) {
	log.debug(msg() + "startingVersions=" + startingVersions.size() + " " + startingVersions);
	}
	}

	// Fire off the requests before getting our own recent updates (for better concurrency)
	// This also allows us to avoid getting updates we don't need... if we got our updates and then got their updates, they would
	// have newer stuff that we also had (assuming updates are going on and are being forwarded).
	for (String replica : replicas) {
	requestVersions(replica);
	}

	recentUpdates = ulog.getRecentUpdates();
	try {
	ourUpdates = recentUpdates.getVersions(nUpdates);
	} finally {
	recentUpdates.close();
	}

	Collections.sort(ourUpdates, absComparator);

	if (startingVersions != null) {
	if (startingVersions.size() == 0) {
	log.warn("no frame of reference to tell if we've missed updates");
	return false;
	}
	Collections.sort(startingVersions, absComparator);

	ourLowThreshold = percentile(startingVersions, 0.8f);
	ourHighThreshold = percentile(startingVersions, 0.2f);

	// now make sure that the starting updates overlap our updates
	// there shouldn't be reorders, so any overlap will do.

	long smallestNewUpdate = Math.abs(ourUpdates.get(ourUpdates.size()-1));

	if (Math.abs(startingVersions.get(0)) < smallestNewUpdate) {
	log.warn(msg() + "too many updates received since start - startingUpdates no longer overlaps with our currentUpdates");
	return false;
	}

	// let's merge the lists
	List<Long> newList = new ArrayList<>(ourUpdates);
	for (Long ver : startingVersions) {
	if (Math.abs(ver) < smallestNewUpdate) {
	newList.add(ver);
	}
	}

	ourUpdates = newList;
	} else {

	if (ourUpdates.size() > 0) {
	ourLowThreshold = percentile(ourUpdates, 0.8f);
	ourHighThreshold = percentile(ourUpdates, 0.2f);
	} else {
	// we have no versions and hence no frame of reference to tell if we can use a peers
	// updates to bring us into sync
	log.info(msg() + "DONE. We have no versions. sync failed.");
	return false;
	}
	}

	ourUpdateSet = new HashSet<>(ourUpdates);
	requestedUpdateSet = new HashSet<>(ourUpdates);

	for(;;) {
	ShardResponse srsp = shardHandler.takeCompletedOrError();
	if (srsp == null) break;
	boolean success = handleResponse(srsp);
	if (!success) {
	log.info(msg() + "DONE. sync failed");
	shardHandler.cancelAll();
	return false;
	}
	}

	log.info(msg() + "DONE. sync succeeded");
	return true;
	}

	private void requestVersions(String replica) {
	SyncShardRequest sreq = new SyncShardRequest();
	sreq.purpose = 1;
	sreq.shards = new String[]{replica};
	sreq.actualShards = sreq.shards;
	sreq.params = new ModifiableSolrParams();
	sreq.params.set("qt","/get");
	sreq.params.set("distrib",false);
	sreq.params.set("getVersions",nUpdates);
	shardHandler.submit(sreq, replica, sreq.params);
	}

	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);
	}
	}

	// sometimes the root exception is a SocketTimeoutException, but ConnectTimeoutException
	// is in the chain
	private boolean connectTimeoutExceptionInChain(Throwable exception) {
	Throwable t = exception;
	while (true) {
	if (t instanceof ConnectTimeoutException) {
	return true;
	}
	Throwable cause = t.getCause();
	if (cause != null) {
	t = cause;
	} else {
	return false;
	}
	}
	}

	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);
	}

	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;
	}


	private boolean handleUpdates(ShardResponse srsp) {
	// we retrieved the last N updates from the replica
	List<Object> updates = (List<Object>)srsp.getSolrResponse().getResponse().get("updates");

	SyncShardRequest sreq = (SyncShardRequest) srsp.getShardRequest();
	if (updates.size() < sreq.requestedUpdates.size()) {
	log.error(msg() + " Requested " + sreq.requestedUpdates.size() + " updates from " + sreq.shards[0] + " but retrieved " + updates.size());
	return false;
	}

	ModifiableSolrParams params = new ModifiableSolrParams();
	params.set(DISTRIB_UPDATE_PARAM, FROMLEADER.toString());
	params.set("peersync",true); // debugging
	SolrQueryRequest req = new LocalSolrQueryRequest(uhandler.core, params);
	SolrQueryResponse rsp = new SolrQueryResponse();

	UpdateRequestProcessorChain processorChain = req.getCore().getUpdateProcessingChain(null);
	UpdateRequestProcessor proc = processorChain.createProcessor(req, rsp);

	Collections.sort(updates, updateRecordComparator);

	Object o = null;
	long lastVersion = 0;
	try {
	// Apply oldest updates first
	for (Object obj : updates) {
	// should currently be a List<Oper,Ver,Doc/Id>
	o = obj;
	List<Object> entry = (List<Object>)o;

	if (debug) {
	log.debug(msg() + "raw update record " + o);
	}

	int oper = (Integer)entry.get(0) & UpdateLog.OPERATION_MASK;
	long version = (Long) entry.get(1);
	if (version == lastVersion && version != 0) continue;
	lastVersion = version;

	switch (oper) {
	case UpdateLog.ADD:
	{
	// byte[] idBytes = (byte[]) entry.get(2);
	SolrInputDocument sdoc = (SolrInputDocument)entry.get(entry.size()-1);
	AddUpdateCommand cmd = new AddUpdateCommand(req);
	// cmd.setIndexedId(new BytesRef(idBytes));
	cmd.solrDoc = sdoc;
	cmd.setVersion(version);
	cmd.setFlags(UpdateCommand.PEER_SYNC \| UpdateCommand.IGNORE_AUTOCOMMIT);
	if (debug) {
	log.debug(msg() + "add " + cmd);
	}
	proc.processAdd(cmd);
	break;
	}
	case UpdateLog.DELETE:
	{
	byte[] idBytes = (byte[]) entry.get(2);
	DeleteUpdateCommand cmd = new DeleteUpdateCommand(req);
	cmd.setIndexedId(new BytesRef(idBytes));
	cmd.setVersion(version);
	cmd.setFlags(UpdateCommand.PEER_SYNC \| UpdateCommand.IGNORE_AUTOCOMMIT);
	if (debug) {
	log.debug(msg() + "delete " + cmd);
	}
	proc.processDelete(cmd);
	break;
	}

	case UpdateLog.DELETE_BY_QUERY:
	{
	String query = (String)entry.get(2);
	DeleteUpdateCommand cmd = new DeleteUpdateCommand(req);
	cmd.query = query;
	cmd.setVersion(version);
	cmd.setFlags(UpdateCommand.PEER_SYNC \| UpdateCommand.IGNORE_AUTOCOMMIT);
	if (debug) {
	log.debug(msg() + "deleteByQuery " + cmd);
	}
	proc.processDelete(cmd);
	break;
	}

	default:
	throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Unknown Operation! " + oper);
	}

	}

	}
	catch (IOException e) {
	// TODO: should this be handled separately as a problem with us?
	// I guess it probably already will by causing replication to be kicked off.
	sreq.updateException = e;
	log.error(msg() + "Error applying updates from " + sreq.shards + " ,update=" + o, e);
	return false;
	}
	catch (Exception e) {
	sreq.updateException = e;
	log.error(msg() + "Error applying updates from " + sreq.shards + " ,update=" + o, e);
	return false;
	}
	finally {
	try {
	proc.finish();
	} catch (Exception e) {
	sreq.updateException = e;
	log.error(msg() + "Error applying updates from " + sreq.shards + " ,finish()", e);
	return false;
	}
	}

	return true;
	}



	/** Requests and applies recent updates from peers */
	public static void sync(SolrCore core, List<String> replicas, int nUpdates) {
	ShardHandlerFactory shardHandlerFactory = core.getCoreDescriptor().getCoreContainer().getShardHandlerFactory();

	ShardHandler shardHandler = shardHandlerFactory.getShardHandler();

	for (String replica : replicas) {
	ShardRequest sreq = new ShardRequest();
	sreq.shards = new String[]{replica};
	sreq.params = new ModifiableSolrParams();
	sreq.params.set("qt","/get");
	sreq.params.set("distrib", false);
	sreq.params.set("getVersions",nUpdates);
	shardHandler.submit(sreq, replica, sreq.params);
	}

	for (String replica : replicas) {
	ShardResponse srsp = shardHandler.takeCompletedOrError();
	}

	}

	}