Index: src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
===================================================================
--- src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java (revision 1481143)
+++ src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java (working copy)
@@ -521,12 +521,16 @@
}
private void connectToPeers() {
+ int sleepMultiplier = 1;
+
// Connect to peer cluster first, unless we have to stop
while (this.isActive() && this.currentPeers.size() == 0) {
try {
chooseSinks();
- Thread.sleep(this.sleepForRetries);
+ if (sleepForRetries("Waiting for peers", sleepMultiplier, true)) {
+ sleepMultiplier++;
+ }
} catch (InterruptedException e) {
LOG.error("Interrupted while trying to connect to sinks", e);
}
@@ -647,10 +651,30 @@
*/
protected boolean sleepForRetries(String msg, int sleepMultiplier) {
try {
+ return sleepForRetries(msg, sleepMultiplier, false);
+ } catch (InterruptedException ignore) {
+ // can't happen
+ throw new RuntimeException(ignore);
+ }
+ }
+ /**
+ * Do the sleeping logic
+ * @param msg Why we sleep
+ * @param sleepMultiplier by how many times the default sleeping time is augmented
+ * @param retrowExceptions throw InterruptedException up to the caller
+ * @return True if sleepMultiplier is < maxRetriesMultiplier
+ */
+ protected boolean sleepForRetries(String msg, int sleepMultiplier, boolean retrowExceptions)
+ throws InterruptedException {
+ try {
LOG.debug(msg + ", sleeping " + sleepForRetries + " times " + sleepMultiplier);
Thread.sleep(this.sleepForRetries * sleepMultiplier);
} catch (InterruptedException e) {
- LOG.debug("Interrupted while sleeping between retries");
+ if (retrowExceptions) {
+ throw e;
+ } else {
+ LOG.debug("Interrupted while sleeping between retries");
+ }
}
return sleepMultiplier < maxRetriesMultiplier;
}