Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-25117 ReplicationSourceShipper thread can not be finished (#2521) 

Signed-off-by: Wellington Chevreuil <wellington.chevreuil@gmail.com>
Signed-off-by: stack <stack@apache.org>
Signed-off-by: Guanghao Zhang <zghao@apache.org>
Signed-off-by: Duo Zhang <zhangduo@apache.org>

(cherry picked from commit 78b7244091)
This commit is contained in:
XinSun 2020-10-15 01:08:54 +08:00 committed by Wellington Chevreuil
parent 5d65bd45fa
commit 4e6c52c319
2 changed files with 6 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java
View File

@ -238,7 +238,7 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi
* @param sleepMultiplier by how many times the default sleeping time is augmented
* @return True if <code>sleepMultiplier</code> is &lt; <code>maxRetriesMultiplier</code>
*/
protected boolean sleepForRetries(String msg, int sleepMultiplier) {
private boolean sleepForRetries(String msg, int sleepMultiplier) {
try {
if (LOG.isTraceEnabled()) {
LOG.trace("{} {}, sleeping {} times {}",
@ -246,8 +246,9 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi
}
Thread.sleep(this.sleepForRetries * sleepMultiplier);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
if (LOG.isDebugEnabled()) {
LOG.debug("{} Interrupted while sleeping between retries", logPeerId());
LOG.debug("{} {} Interrupted while sleeping between retries", msg, logPeerId());
}
}
return sleepMultiplier < maxRetriesMultiplier;

6
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
View File

@ -714,6 +714,9 @@ public class ReplicationSource implements ReplicationSourceInterface {
}
}
if (this.replicationEndpoint != null) {
this.replicationEndpoint.stop();
}
for (ReplicationSourceShipper worker : workers) {
if (worker.isAlive() || worker.entryReader.isAlive()) {
try {
@ -734,9 +737,6 @@ public class ReplicationSource implements ReplicationSourceInterface {
}
}
if (this.replicationEndpoint != null) {
this.replicationEndpoint.stop();
}
if (join) {
for (ReplicationSourceShipper worker : workers) {
Threads.shutdown(worker, this.sleepForRetries);