Return-Path: X-Original-To: apmail-hbase-commits-archive@www.apache.org Delivered-To: apmail-hbase-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id A104518863 for ; Fri, 11 Dec 2015 21:21:36 +0000 (UTC) Received: (qmail 65408 invoked by uid 500); 11 Dec 2015 21:21:36 -0000 Delivered-To: apmail-hbase-commits-archive@hbase.apache.org Received: (qmail 65364 invoked by uid 500); 11 Dec 2015 21:21:36 -0000 Mailing-List: contact commits-help@hbase.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@hbase.apache.org Delivered-To: mailing list commits@hbase.apache.org Received: (qmail 65355 invoked by uid 99); 11 Dec 2015 21:21:36 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 11 Dec 2015 21:21:36 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 62098E0901; Fri, 11 Dec 2015 21:21:36 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: eclark@apache.org To: commits@hbase.apache.org Message-Id: X-Mailer: ASF-Git Admin Mailer Subject: hbase git commit: HBASE-14953 Replication: retry on RejectedExecutionException Date: Fri, 11 Dec 2015 21:21:36 +0000 (UTC) Repository: hbase Updated Branches: refs/heads/master 22b95aebc -> 453a66c3b 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 Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/453a66c3 Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/453a66c3 Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/453a66c3 Branch: refs/heads/master Commit: 453a66c3b9d4f51311db6acc5cba84290dbd29ed Parents: 22b95ae Author: Ashu Pachauri Authored: Tue Dec 8 14:25:41 2015 -0800 Committer: Elliott Clark Committed: Fri Dec 11 13:15:22 2015 -0800 ---------------------------------------------------------------------- .../regionserver/HBaseInterClusterReplicationEndpoint.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/453a66c3/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java ---------------------------------------------------------------------- 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 d51d512..70cc420 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,