From 48d8e996ed94da21970a9e0dba90e7ba576e1698 Mon Sep 17 00:00:00 2001
From: Michael Stack <stack@apache.org>
Date: Tue, 25 Feb 2014 06:31:10 +0000
Subject: [PATCH] HBASE-10575 ReplicationSource thread can't be terminated if
 it runs into the loop to contact peer's zk ensemble and fails continuously

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1571579 13f79535-47bb-0310-9956-ffa450edef68
---
 .../regionserver/ReplicationSource.java       | 35 ++++++++++++-------
 1 file changed, 23 insertions(+), 12 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
index fe9451592ea..673096014ef 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
@@ -207,24 +207,43 @@ public class ReplicationSource extends Thread
     }
   }
 
+  private void uninitialize() {
+    if (this.conn != null) {
+      try {
+        this.conn.close();
+      } catch (IOException e) {
+        LOG.debug("Attempt to close connection failed", e);
+      }
+    }
+    LOG.debug("Source exiting " + this.peerId);
+    metrics.clear();
+  }
+
   @Override
   public void run() {
     connectToPeers();
     // We were stopped while looping to connect to sinks, just abort
     if (!this.isActive()) {
-      metrics.clear();
+      uninitialize();
       return;
     }
+
     int sleepMultiplier = 1;
     // delay this until we are in an asynchronous thread
-    while (this.peerClusterId == null) {
+    while (this.isActive() && this.peerClusterId == null) {
       this.peerClusterId = replicationPeers.getPeerUUID(this.peerId);
-      if (this.peerClusterId == null) {
+      if (this.isActive() && this.peerClusterId == null) {
         if (sleepForRetries("Cannot contact the peer's zk ensemble", sleepMultiplier)) {
           sleepMultiplier++;
         }
       }
     }
+    // We were stopped while looping to contact peer's zk ensemble, just abort
+    if (!this.isActive()) {
+      uninitialize();
+      return;
+    }
+
     // resetting to 1 to reuse later
     sleepMultiplier = 1;
 
@@ -365,15 +384,7 @@ public class ReplicationSource extends Thread
       sleepMultiplier = 1;
       shipEdits(currentWALisBeingWrittenTo, entries);
     }
-    if (this.conn != null) {
-      try {
-        this.conn.close();
-      } catch (IOException e) {
-        LOG.debug("Attempt to close connection failed", e);
-      }
-    }
-    LOG.debug("Source exiting " + this.peerId);
-    metrics.clear();
+    uninitialize();
   }
 
   /**