svn commit: r1485318 - /hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java

[la...@apache.org]

Author: larsh
Date: Wed May 22 18:01:17 2013
New Revision: 1485318

URL: http://svn.apache.org/r1485318
Log:
HBASE-8525 Use sleep multilier when choosing sinks in ReplicationSource

Modified:
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java?rev=1485318&r1=1485317&r2=1485318&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java Wed May 22 18:01:17 2013
@@ -171,9 +171,9 @@ public class ReplicationSource extends T
     for (int i = 0; i < this.replicationQueueNbCapacity; i++) {
       this.entriesArray[i] = new HLog.Entry();
     }
-    this.maxRetriesMultiplier =
-        this.conf.getInt("replication.source.maxretriesmultiplier", 10);
-    this.socketTimeoutMultiplier = maxRetriesMultiplier * maxRetriesMultiplier;
+    this.maxRetriesMultiplier = this.conf.getInt("replication.source.maxretriesmultiplier", 10);
+    this.socketTimeoutMultiplier = this.conf.getInt("replication.source.socketTimeoutMultiplier",
+        maxRetriesMultiplier * maxRetriesMultiplier);
     this.queue =
         new PriorityBlockingQueue<Path>(
             conf.getInt("hbase.regionserver.maxlogs", 32),
@@ -521,14 +521,16 @@ public class ReplicationSource extends T
   }
 
   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);
-      } catch (InterruptedException e) {
-        LOG.error("Interrupted while trying to connect to sinks", e);
+      chooseSinks();
+      if (this.isActive() && this.currentPeers.size() == 0) {
+        if (sleepForRetries("Waiting for peers", sleepMultiplier)) {
+          sleepMultiplier++;
+        }
       }
     }
   }