From 453a66c3b9d4f51311db6acc5cba84290dbd29ed Mon Sep 17 00:00:00 2001 From: Ashu Pachauri Date: Tue, 8 Dec 2015 14:25:41 -0800 Subject: [PATCH] HBASE-14953 Replication: retry on RejectedExecutionException In HBaseInterClusterReplicationEndpoint, we fail the whole batch in case of a RejectedExecutionException on an individual sub-batch. We should let the submitted sub-batches finish and retry only for the remaining ones. Signed-off-by: Elliott Clark --- .../regionserver/HBaseInterClusterReplicationEndpoint.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java index d51d512471e..70cc4204248 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java @@ -29,7 +29,7 @@ import java.util.concurrent.CompletionService; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorCompletionService; import java.util.concurrent.Future; -import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -113,8 +113,9 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi // per sink thread pool this.maxThreads = this.conf.getInt(HConstants.REPLICATION_SOURCE_MAXTHREADS_KEY, HConstants.REPLICATION_SOURCE_MAXTHREADS_DEFAULT); - this.exec = new ThreadPoolExecutor(1, maxThreads, 60, TimeUnit.SECONDS, - new SynchronousQueue()); + this.exec = new ThreadPoolExecutor(maxThreads, maxThreads, 60, TimeUnit.SECONDS, + new LinkedBlockingQueue()); + this.exec.allowCoreThreadTimeOut(true); this.replicationBulkLoadDataEnabled = conf.getBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY,