You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by vj...@apache.org on 2020/08/12 10:28:13 UTC
[hbase] branch branch-2 updated: HBASE-24750 : All ExecutorService
should use guava ThreadFactoryBuilder (#2214)
This is an automated email from the ASF dual-hosted git repository.
vjasani pushed a commit to branch branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2 by this push:
new 8ccf643 HBASE-24750 : All ExecutorService should use guava ThreadFactoryBuilder (#2214)
8ccf643 is described below
commit 8ccf643fdc96a5f70106a50fb08f0e37dcf80ec5
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Wed Aug 12 15:57:53 2020 +0530
HBASE-24750 : All ExecutorService should use guava ThreadFactoryBuilder (#2214)
Closes #2196
Signed-off-by: Nick Dimiduk <nd...@apache.org>
Signed-off-by: Ted Yu <ty...@apache.org>
Signed-off-by: niuyulin <ny...@163.com>
---
.../hadoop/hbase/client/AsyncConnectionImpl.java | 5 +-
.../hadoop/hbase/client/ClusterStatusListener.java | 6 +-
.../hbase/client/ConnectionImplementation.java | 12 ++--
.../org/apache/hadoop/hbase/client/HTable.java | 9 ++-
.../apache/hadoop/hbase/ipc/AbstractRpcClient.java | 9 ++-
.../hadoop/hbase/ipc/NettyRpcConnection.java | 6 +-
.../hadoop/hbase/client/TestAsyncProcess.java | 5 +-
.../hadoop/hbase/client/TestClientNoCluster.java | 4 +-
.../java/org/apache/hadoop/hbase/util/Threads.java | 84 ++--------------------
.../hbase/client/example/AsyncClientExample.java | 4 +-
.../procedure2/RemoteProcedureDispatcher.java | 5 +-
.../hbase/master/ClusterStatusPublisher.java | 6 +-
.../assignment/SplitTableRegionProcedure.java | 4 +-
.../java/org/apache/hadoop/hbase/mob/MobUtils.java | 24 +++----
.../hbase/namequeues/NamedQueueRecorder.java | 4 +-
.../hadoop/hbase/regionserver/MemStoreFlusher.java | 6 +-
.../hadoop/hbase/regionserver/wal/FSHLog.java | 7 +-
.../RegionReplicaReplicationEndpoint.java | 13 ++--
.../hadoop/hbase/snapshot/SnapshotManifest.java | 5 +-
.../hadoop/hbase/tool/HFileContentValidator.java | 4 +-
.../java/org/apache/hadoop/hbase/util/FSUtils.java | 4 +-
.../org/apache/hadoop/hbase/util/HBaseFsck.java | 5 +-
.../hadoop/hbase/util/ModifyRegionUtils.java | 4 +-
.../apache/hadoop/hbase/util/MultiHConnection.java | 6 +-
.../org/apache/hadoop/hbase/wal/OutputSink.java | 4 +-
.../hadoop/hbase/AcidGuaranteesTestTool.java | 7 +-
.../client/TestAsyncTableGetMultiThreaded.java | 6 +-
.../hbase/client/TestConnectionImplementation.java | 9 +--
.../coprocessor/TestOpenTableInCoprocessor.java | 6 +-
.../hbase/master/assignment/TestRegionStates.java | 12 ++--
.../hbase/mob/compactions/TestMobCompactor.java | 24 +++----
.../compactions/TestPartitionedMobCompactor.java | 17 +++--
.../hbase/procedure/SimpleRSProcedureManager.java | 5 +-
.../TestRegionServerReportForDuty.java | 6 +-
.../hbase/regionserver/wal/TestAsyncFSWAL.java | 5 +-
.../hbase/regionserver/wal/TestAsyncWALReplay.java | 5 +-
.../apache/hadoop/hbase/util/TestHBaseFsckMOB.java | 7 +-
.../hadoop/hbase/thrift/IncrementCoalescer.java | 7 +-
.../apache/hadoop/hbase/zookeeper/ZKWatcher.java | 6 +-
39 files changed, 176 insertions(+), 191 deletions(-)
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
index 9f935f0..039ff4d 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
@@ -51,6 +51,7 @@ import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.ConcurrentMapUtils;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,7 +74,9 @@ class AsyncConnectionImpl implements AsyncConnection {
@VisibleForTesting
static final HashedWheelTimer RETRY_TIMER = new HashedWheelTimer(
- Threads.newDaemonThreadFactory("Async-Client-Retry-Timer"), 10, TimeUnit.MILLISECONDS);
+ new ThreadFactoryBuilder().setNameFormat("Async-Client-Retry-Timer-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(), 10,
+ TimeUnit.MILLISECONDS);
private static final String RESOLVE_HOSTNAME_ON_FAIL_KEY = "hbase.resolve.hostnames.on.failure";
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterStatusListener.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterStatusListener.java
index ccdfec7..a7e715c 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterStatusListener.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterStatusListener.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.util.Addressing;
import org.apache.hadoop.hbase.util.ExceptionUtil;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hbase.thirdparty.io.netty.bootstrap.Bootstrap;
import org.apache.hbase.thirdparty.io.netty.buffer.ByteBufInputStream;
import org.apache.hbase.thirdparty.io.netty.channel.ChannelHandlerContext;
@@ -178,8 +179,9 @@ class ClusterStatusListener implements Closeable {
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
class MulticastListener implements Listener {
private DatagramChannel channel;
- private final EventLoopGroup group = new NioEventLoopGroup(
- 1, Threads.newDaemonThreadFactory("hbase-client-clusterStatusListener"));
+ private final EventLoopGroup group = new NioEventLoopGroup(1,
+ new ThreadFactoryBuilder().setNameFormat("hbase-client-clusterStatusListener-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
public MulticastListener() {
}
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java
index 06e243b..b3b7b7d 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java
@@ -79,6 +79,7 @@ import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
@@ -490,13 +491,10 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
coreThreads = maxThreads;
}
- ThreadPoolExecutor tpe = new ThreadPoolExecutor(
- coreThreads,
- maxThreads,
- keepAliveTime,
- TimeUnit.SECONDS,
- workQueue,
- Threads.newDaemonThreadFactory(toString() + nameHint));
+ ThreadPoolExecutor tpe =
+ new ThreadPoolExecutor(coreThreads, maxThreads, keepAliveTime, TimeUnit.SECONDS, workQueue,
+ new ThreadFactoryBuilder().setNameFormat(toString() + nameHint + "-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
tpe.allowCoreThreadTimeOut(true);
return tpe;
}
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java
index 51e45b1..a6866d2 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java
@@ -34,6 +34,8 @@ import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.io.TimeRange;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
import org.slf4j.Logger;
@@ -56,7 +58,6 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.RegionActi
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.ReflectionUtils;
-import org.apache.hadoop.hbase.util.Threads;
import java.io.IOException;
import java.io.InterruptedIOException;
@@ -138,8 +139,10 @@ public class HTable implements Table {
// if it is necessary and will grow unbounded. This could be bad but in HCM
// we only create as many Runnables as there are region servers. It means
// it also scales when new region servers are added.
- ThreadPoolExecutor pool = new ThreadPoolExecutor(corePoolSize, maxThreads, keepAliveTime,
- TimeUnit.SECONDS, new SynchronousQueue<>(), Threads.newDaemonThreadFactory("htable"));
+ ThreadPoolExecutor pool =
+ new ThreadPoolExecutor(corePoolSize, maxThreads, keepAliveTime, TimeUnit.SECONDS,
+ new SynchronousQueue<>(), new ThreadFactoryBuilder().setNameFormat("htable-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
pool.allowCoreThreadTimeOut(true);
return pool;
}
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java
index bf2f361..064664e 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java
@@ -44,6 +44,7 @@ import org.apache.hadoop.hbase.util.PoolMap;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.ipc.RemoteException;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -91,10 +92,14 @@ public abstract class AbstractRpcClient<T extends RpcConnection> implements RpcC
public static final Logger LOG = LoggerFactory.getLogger(AbstractRpcClient.class);
protected static final HashedWheelTimer WHEEL_TIMER = new HashedWheelTimer(
- Threads.newDaemonThreadFactory("RpcClient-timer"), 10, TimeUnit.MILLISECONDS);
+ new ThreadFactoryBuilder().setNameFormat("RpcClient-timer-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(), 10,
+ TimeUnit.MILLISECONDS);
private static final ScheduledExecutorService IDLE_CONN_SWEEPER = Executors
- .newScheduledThreadPool(1, Threads.newDaemonThreadFactory("Idle-Rpc-Conn-Sweeper"));
+ .newScheduledThreadPool(1,
+ new ThreadFactoryBuilder().setNameFormat("Idle-Rpc-Conn-Sweeper-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
protected boolean running = true; // if client runs
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java
index 7c9d927..3fd6ac9 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.hbase.security.NettyHBaseSaslRpcClientHandler;
import org.apache.hadoop.hbase.security.SaslChallengeDecoder;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,8 +77,9 @@ class NettyRpcConnection extends RpcConnection {
private static final Logger LOG = LoggerFactory.getLogger(NettyRpcConnection.class);
- private static final ScheduledExecutorService RELOGIN_EXECUTOR =
- Executors.newSingleThreadScheduledExecutor(Threads.newDaemonThreadFactory("Relogin"));
+ private static final ScheduledExecutorService RELOGIN_EXECUTOR = Executors
+ .newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("Relogin-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
private final NettyRpcClient rpcClient;
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
index f750fd6..f4a27fe 100644
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
@@ -68,6 +68,7 @@ import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.Assert;
import org.junit.Before;
import org.junit.ClassRule;
@@ -138,7 +139,9 @@ public class TestAsyncProcess {
static class CountingThreadFactory implements ThreadFactory {
final AtomicInteger nbThreads;
- ThreadFactory realFactory = Threads.newDaemonThreadFactory("test-TestAsyncProcess");
+ ThreadFactory realFactory =
+ new ThreadFactoryBuilder().setNameFormat("test-TestAsyncProcess-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build();
@Override
public Thread newThread(Runnable r) {
nbThreads.incrementAndGet();
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientNoCluster.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientNoCluster.java
index 4150484..f1090f5 100644
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientNoCluster.java
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientNoCluster.java
@@ -59,6 +59,7 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Ignore;
@@ -803,7 +804,8 @@ public class TestClientNoCluster extends Configured implements Tool {
// Have them all share the same connection so they all share the same instance of
// ManyServersManyRegionsConnection so I can keep an eye on how many requests by server.
final ExecutorService pool = Executors.newCachedThreadPool(
- Threads.newDaemonThreadFactory("p"));
+ new ThreadFactoryBuilder().setNameFormat("p-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
// Executors.newFixedThreadPool(servers * 10, Threads.getNamedThreadFactory("p"));
// Share a connection so I can keep counts in the 'server' on concurrency.
final Connection sharedConnection = ConnectionFactory.createConnection(getConf()/*, pool*/);
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Threads.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Threads.java
index 3527340..ab6f805 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Threads.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Threads.java
@@ -45,16 +45,9 @@ import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
@InterfaceAudience.Private
public class Threads {
private static final Logger LOG = LoggerFactory.getLogger(Threads.class);
- private static final AtomicInteger poolNumber = new AtomicInteger(1);
public static final UncaughtExceptionHandler LOGGING_EXCEPTION_HANDLER =
- new UncaughtExceptionHandler() {
- @Override
- public void uncaughtException(Thread t, Throwable e) {
- LOG.warn("Thread:" + t + " exited with Exception:"
- + StringUtils.stringifyException(e));
- }
- };
+ (t, e) -> LOG.warn("Thread:{} exited with Exception:{}", t, StringUtils.stringifyException(e));
/**
* Utility method that sets name, daemon status and starts passed thread.
@@ -186,81 +179,16 @@ public class Threads {
* @return threadPoolExecutor the cachedThreadPool with a bounded number
* as the maximum thread size in the pool.
*/
- public static ThreadPoolExecutor getBoundedCachedThreadPool(
- int maxCachedThread, long timeout, TimeUnit unit,
- ThreadFactory threadFactory) {
+ public static ThreadPoolExecutor getBoundedCachedThreadPool(int maxCachedThread, long timeout,
+ TimeUnit unit, ThreadFactory threadFactory) {
ThreadPoolExecutor boundedCachedThreadPool =
- new ThreadPoolExecutor(maxCachedThread, maxCachedThread, timeout,
- unit, new LinkedBlockingQueue<>(), threadFactory);
+ new ThreadPoolExecutor(maxCachedThread, maxCachedThread, timeout, unit,
+ new LinkedBlockingQueue<>(), threadFactory);
// allow the core pool threads timeout and terminate
boundedCachedThreadPool.allowCoreThreadTimeOut(true);
return boundedCachedThreadPool;
}
-
- /**
- * Returns a {@link java.util.concurrent.ThreadFactory} that names each created thread uniquely,
- * with a common prefix.
- * @param prefix The prefix of every created Thread's name
- * @return a {@link java.util.concurrent.ThreadFactory} that names threads
- */
- public static ThreadFactory getNamedThreadFactory(final String prefix) {
- SecurityManager s = System.getSecurityManager();
- final ThreadGroup threadGroup = (s != null) ? s.getThreadGroup() : Thread.currentThread()
- .getThreadGroup();
-
- return new ThreadFactory() {
- final AtomicInteger threadNumber = new AtomicInteger(1);
- private final int poolNumber = Threads.poolNumber.getAndIncrement();
- final ThreadGroup group = threadGroup;
-
- @Override
- public Thread newThread(Runnable r) {
- final String name = prefix + "-pool" + poolNumber + "-t" + threadNumber.getAndIncrement();
- return new Thread(group, r, name);
- }
- };
- }
-
- /**
- * Same as {#newDaemonThreadFactory(String, UncaughtExceptionHandler)},
- * without setting the exception handler.
- */
- public static ThreadFactory newDaemonThreadFactory(final String prefix) {
- return newDaemonThreadFactory(prefix, null);
- }
-
- /**
- * Get a named {@link ThreadFactory} that just builds daemon threads.
- * @param prefix name prefix for all threads created from the factory
- * @param handler unhandles exception handler to set for all threads
- * @return a thread factory that creates named, daemon threads with
- * the supplied exception handler and normal priority
- */
- public static ThreadFactory newDaemonThreadFactory(final String prefix,
- final UncaughtExceptionHandler handler) {
- final ThreadFactory namedFactory = getNamedThreadFactory(prefix);
- return new ThreadFactory() {
- @Override
- public Thread newThread(Runnable r) {
- Thread t = namedFactory.newThread(r);
- if (handler != null) {
- t.setUncaughtExceptionHandler(handler);
- } else {
- t.setUncaughtExceptionHandler(LOGGING_EXCEPTION_HANDLER);
- }
- if (!t.isDaemon()) {
- t.setDaemon(true);
- }
- if (t.getPriority() != Thread.NORM_PRIORITY) {
- t.setPriority(Thread.NORM_PRIORITY);
- }
- return t;
- }
-
- };
- }
-
/** Sets an UncaughtExceptionHandler for the thread which logs the
* Exception stack if the thread dies.
*/
@@ -268,7 +196,7 @@ public class Threads {
t.setUncaughtExceptionHandler(LOGGING_EXCEPTION_HANDLER);
}
- private static interface PrintThreadInfoHelper {
+ private interface PrintThreadInfoHelper {
void printThreadInfo(PrintStream stream, String title);
diff --git a/hbase-examples/src/main/java/org/apache/hadoop/hbase/client/example/AsyncClientExample.java b/hbase-examples/src/main/java/org/apache/hadoop/hbase/client/example/AsyncClientExample.java
index b8b3213..8e34af9 100644
--- a/hbase-examples/src/main/java/org/apache/hadoop/hbase/client/example/AsyncClientExample.java
+++ b/hbase-examples/src/main/java/org/apache/hadoop/hbase/client/example/AsyncClientExample.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -130,7 +131,8 @@ public class AsyncClientExample extends Configured implements Tool {
TableName tableName = TableName.valueOf(args[0]);
int numOps = args.length > 1 ? Integer.parseInt(args[1]) : DEFAULT_NUM_OPS;
ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_POOL_SIZE,
- Threads.newDaemonThreadFactory("AsyncClientExample"));
+ new ThreadFactoryBuilder().setNameFormat("AsyncClientExample-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
// We use AsyncTable here so we need to provide a separated thread pool. RawAsyncTable does not
// need a thread pool and may have a better performance if you use it correctly as it can save
// some context switches. But if you use RawAsyncTable incorrectly, you may have a very bad
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java
index ecb4ffc..dd76083 100644
--- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.hbase.procedure2.util.DelayedUtil.DelayedWithTimeout;
import org.apache.hadoop.hbase.procedure2.util.StringUtils;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -100,8 +101,8 @@ public abstract class RemoteProcedureDispatcher<TEnv, TRemote extends Comparable
// Create the thread pool that will execute RPCs
threadPool = Threads.getBoundedCachedThreadPool(corePoolSize, 60L, TimeUnit.SECONDS,
- Threads.newDaemonThreadFactory(this.getClass().getSimpleName(),
- getUncaughtExceptionHandler()));
+ new ThreadFactoryBuilder().setNameFormat(this.getClass().getSimpleName() + "-pool-%d")
+ .setUncaughtExceptionHandler(getUncaughtExceptionHandler()).build());
return true;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ClusterStatusPublisher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ClusterStatusPublisher.java
index 8257466..608f118 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ClusterStatusPublisher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ClusterStatusPublisher.java
@@ -47,6 +47,7 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.util.VersionInfo;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.hbase.thirdparty.io.netty.bootstrap.Bootstrap;
import org.apache.hbase.thirdparty.io.netty.buffer.Unpooled;
@@ -245,8 +246,9 @@ public class ClusterStatusPublisher extends ScheduledChore {
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
public static class MulticastPublisher implements Publisher {
private DatagramChannel channel;
- private final EventLoopGroup group = new NioEventLoopGroup(
- 1, Threads.newDaemonThreadFactory("hbase-master-clusterStatusPublisher"));
+ private final EventLoopGroup group = new NioEventLoopGroup(1,
+ new ThreadFactoryBuilder().setNameFormat("hbase-master-clusterStatusPublisher-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
public MulticastPublisher() {
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java
index c9cc906..2d91c55 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java
@@ -71,6 +71,7 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.WALSplitUtil;
import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -679,7 +680,8 @@ public class SplitTableRegionProcedure
LOG.info("pid=" + getProcId() + " splitting " + nbFiles + " storefiles, region=" +
getParentRegion().getShortNameToLog() + ", threads=" + maxThreads);
final ExecutorService threadPool = Executors.newFixedThreadPool(maxThreads,
- Threads.newDaemonThreadFactory("StoreFileSplitter-%1$d"));
+ new ThreadFactoryBuilder().setNameFormat("StoreFileSplitter-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
final List<Future<Pair<Path, Path>>> futures = new ArrayList<Future<Pair<Path, Path>>>(nbFiles);
// Split each store file.
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java
index 2a6e637..a957888 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/MobUtils.java
@@ -31,7 +31,6 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -74,6 +73,7 @@ import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -868,18 +868,16 @@ public final class MobUtils {
}
final SynchronousQueue<Runnable> queue = new SynchronousQueue<>();
ThreadPoolExecutor pool = new ThreadPoolExecutor(1, maxThreads, 60, TimeUnit.SECONDS, queue,
- Threads.newDaemonThreadFactory("MobCompactor"), new RejectedExecutionHandler() {
- @Override
- public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
- try {
- // waiting for a thread to pick up instead of throwing exceptions.
- queue.put(r);
- } catch (InterruptedException e) {
- throw new RejectedExecutionException(e);
- }
- }
- });
- ((ThreadPoolExecutor) pool).allowCoreThreadTimeOut(true);
+ new ThreadFactoryBuilder().setNameFormat("MobCompactor-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(), (r, executor) -> {
+ try {
+ // waiting for a thread to pick up instead of throwing exceptions.
+ queue.put(r);
+ } catch (InterruptedException e) {
+ throw new RejectedExecutionException(e);
+ }
+ });
+ pool.allowCoreThreadTimeOut(true);
return pool;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/NamedQueueRecorder.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/NamedQueueRecorder.java
index cb3512a..b5e743e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/NamedQueueRecorder.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/NamedQueueRecorder.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.namequeues.request.NamedQueueGetRequest;
import org.apache.hadoop.hbase.namequeues.response.NamedQueueGetResponse;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
@@ -64,7 +65,8 @@ public class NamedQueueRecorder {
// disruptor initialization with BlockingWaitStrategy
this.disruptor = new Disruptor<>(RingBufferEnvelope::new,
getEventCount(eventCount),
- Threads.newDaemonThreadFactory(hostingThreadName + ".slowlog.append"),
+ new ThreadFactoryBuilder().setNameFormat(hostingThreadName + ".slowlog.append-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(),
ProducerType.MULTI,
new BlockingWaitStrategy());
this.disruptor.setDefaultExceptionHandler(new DisruptorExceptionHandler());
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
index 1537f7c..744bbbe 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
@@ -49,6 +49,7 @@ import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.htrace.core.TraceScope;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
@@ -512,8 +513,9 @@ class MemStoreFlusher implements FlushRequester {
}
synchronized void start(UncaughtExceptionHandler eh) {
- ThreadFactory flusherThreadFactory = Threads.newDaemonThreadFactory(
- server.getServerName().toShortString() + "-MemStoreFlusher", eh);
+ ThreadFactory flusherThreadFactory = new ThreadFactoryBuilder()
+ .setNameFormat(server.getServerName().toShortString() + "-MemStoreFlusher-pool-%d")
+ .setUncaughtExceptionHandler(eh).build();
for (int i = 0; i < flushHandlers.length; i++) {
flushHandlers[i] = new FlushHandler("MemStoreFlusher." + i);
flusherThreadFactory.newThread(flushHandlers[i]);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java
index fb2abfa..faf32ed 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java
@@ -242,9 +242,10 @@ public class FSHLog extends AbstractFSWAL<Writer> {
// Using BlockingWaitStrategy. Stuff that is going on here takes so long it makes no sense
// spinning as other strategies do.
this.disruptor = new Disruptor<>(RingBufferTruck::new,
- getPreallocatedEventCount(),
- Threads.newDaemonThreadFactory(hostingThreadName + ".append"),
- ProducerType.MULTI, new BlockingWaitStrategy());
+ getPreallocatedEventCount(),
+ new ThreadFactoryBuilder().setNameFormat(hostingThreadName + ".append-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(),
+ ProducerType.MULTI, new BlockingWaitStrategy());
// Advance the ring buffer sequence so that it starts from 1 instead of 0,
// because SyncFuture.NOT_DONE = 0.
this.disruptor.getRingBuffer().next();
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java
index cead808..7eca732 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java
@@ -65,6 +65,7 @@ import org.apache.hadoop.hbase.wal.OutputSink;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALSplitter.PipelineController;
import org.apache.hadoop.util.StringUtils;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -198,13 +199,11 @@ public class RegionReplicaReplicationEndpoint extends HBaseReplicationEndpoint {
new LinkedBlockingQueue<>(maxThreads *
conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
- ThreadPoolExecutor tpe = new ThreadPoolExecutor(
- maxThreads,
- maxThreads,
- keepAliveTime,
- TimeUnit.SECONDS,
- workQueue,
- Threads.newDaemonThreadFactory(this.getClass().getSimpleName() + "-rpc-shared-"));
+ ThreadPoolExecutor tpe =
+ new ThreadPoolExecutor(maxThreads, maxThreads, keepAliveTime, TimeUnit.SECONDS, workQueue,
+ new ThreadFactoryBuilder()
+ .setNameFormat(this.getClass().getSimpleName() + "-rpc-shared-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
tpe.allowCoreThreadTimeOut(true);
return tpe;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/SnapshotManifest.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/SnapshotManifest.java
index 6a21581..b6eb7a7 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/SnapshotManifest.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/SnapshotManifest.java
@@ -26,7 +26,6 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ThreadPoolExecutor;
@@ -52,6 +51,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.FSTableDescriptors;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -616,7 +616,8 @@ public final class SnapshotManifest {
public static ThreadPoolExecutor createExecutor(final Configuration conf, final String name) {
int maxThreads = conf.getInt("hbase.snapshot.thread.pool.max", 8);
return Threads.getBoundedCachedThreadPool(maxThreads, 30L, TimeUnit.SECONDS,
- Threads.newDaemonThreadFactory(name));
+ new ThreadFactoryBuilder().setNameFormat(name + "-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
}
/**
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/HFileContentValidator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/HFileContentValidator.java
index 3120d8a..53f0302 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/HFileContentValidator.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/HFileContentValidator.java
@@ -34,6 +34,7 @@ import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.util.hbck.HFileCorruptionChecker;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -108,7 +109,8 @@ public class HFileContentValidator extends AbstractHBaseTool {
int availableProcessors = Runtime.getRuntime().availableProcessors();
int numThreads = conf.getInt("hfilevalidator.numthreads", availableProcessors);
return Executors.newFixedThreadPool(numThreads,
- Threads.newDaemonThreadFactory("hfile-validator"));
+ new ThreadFactoryBuilder().setNameFormat("hfile-validator-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
}
@Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java
index f19f864..3b22de4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java
@@ -84,6 +84,7 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.StringUtils;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1637,7 +1638,8 @@ public final class FSUtils {
// run in multiple threads
final ExecutorService tpe = Executors.newFixedThreadPool(threadPoolSize,
- Threads.newDaemonThreadFactory("FSRegionQuery"));
+ new ThreadFactoryBuilder().setNameFormat("FSRegionQuery-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
try {
// ignore all file status items that are not of interest
for (FileStatus regionStatus : statusList) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
index 1783d7d..a58c681 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
@@ -133,6 +133,7 @@ import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
import org.apache.zookeeper.KeeperException;
@@ -349,7 +350,9 @@ public class HBaseFsck extends Configured implements Closeable {
private static ExecutorService createThreadPool(Configuration conf) {
int numThreads = conf.getInt("hbasefsck.numthreads", MAX_NUM_THREADS);
- return new ScheduledThreadPoolExecutor(numThreads, Threads.newDaemonThreadFactory("hbasefsck"));
+ return new ScheduledThreadPoolExecutor(numThreads,
+ new ThreadFactoryBuilder().setNameFormat("hbasefsck-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
}
/**
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ModifyRegionUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ModifyRegionUtils.java
index c0189c8..99691c8 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ModifyRegionUtils.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ModifyRegionUtils.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -232,7 +233,8 @@ public abstract class ModifyRegionUtils {
"hbase.hregion.open.and.init.threads.max", 16));
ThreadPoolExecutor regionOpenAndInitThreadPool = Threads.
getBoundedCachedThreadPool(maxThreads, 30L, TimeUnit.SECONDS,
- Threads.newDaemonThreadFactory(threadNamePrefix));
+ new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
return regionOpenAndInitThreadPool;
}
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/MultiHConnection.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/MultiHConnection.java
index 5805793..58bd4dd 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/MultiHConnection.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/MultiHConnection.java
@@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -137,8 +138,9 @@ public class MultiHConnection {
* conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
ThreadPoolExecutor tpe =
- new ThreadPoolExecutor(maxThreads, maxThreads, keepAliveTime, TimeUnit.SECONDS, workQueue,
- Threads.newDaemonThreadFactory("MultiHConnection" + "-shared-"));
+ new ThreadPoolExecutor(maxThreads, maxThreads, keepAliveTime, TimeUnit.SECONDS, workQueue,
+ new ThreadFactoryBuilder().setNameFormat("MultiHConnection" + "-shared-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
tpe.allowCoreThreadTimeOut(true);
this.batchPool = tpe;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/OutputSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/OutputSink.java
index 60a190a..826febe 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/OutputSink.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/OutputSink.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.util.CancelableProgressable;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,7 +77,8 @@ public abstract class OutputSink {
this.controller = controller;
this.entryBuffers = entryBuffers;
this.closeThreadPool = Threads.getBoundedCachedThreadPool(numThreads, 30L, TimeUnit.SECONDS,
- Threads.newDaemonThreadFactory("split-log-closeStream-"));
+ new ThreadFactoryBuilder().setNameFormat("split-log-closeStream-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
this.closeCompletionService = new ExecutorCompletionService<>(closeThreadPool);
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/AcidGuaranteesTestTool.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/AcidGuaranteesTestTool.java
index 4564806..a58648f 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/AcidGuaranteesTestTool.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/AcidGuaranteesTestTool.java
@@ -47,6 +47,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.ToolRunner;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -90,8 +91,10 @@ public class AcidGuaranteesTestTool extends AbstractHBaseTool {
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<Runnable>(
maxThreads * HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS);
- ThreadPoolExecutor tpe = new ThreadPoolExecutor(coreThreads, maxThreads, keepAliveTime,
- TimeUnit.SECONDS, workQueue, Threads.newDaemonThreadFactory(toString() + "-shared"));
+ ThreadPoolExecutor tpe =
+ new ThreadPoolExecutor(coreThreads, maxThreads, keepAliveTime, TimeUnit.SECONDS, workQueue,
+ new ThreadFactoryBuilder().setNameFormat(toString() + "-shared-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
tpe.allowCoreThreadTimeOut(true);
return tpe;
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableGetMultiThreaded.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableGetMultiThreaded.java
index 94adce7..6d89299 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableGetMultiThreaded.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableGetMultiThreaded.java
@@ -49,6 +49,7 @@ import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.RetryCounter;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -137,8 +138,9 @@ public class TestAsyncTableGetMultiThreaded {
LOG.info("====== Test started ======");
int numThreads = 7;
AtomicBoolean stop = new AtomicBoolean(false);
- ExecutorService executor =
- Executors.newFixedThreadPool(numThreads, Threads.newDaemonThreadFactory("TestAsyncGet-"));
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads,
+ new ThreadFactoryBuilder().setNameFormat("TestAsyncGet-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
List<Future<?>> futures = new ArrayList<>();
IntStream.range(0, numThreads).forEach(i -> futures.add(executor.submit(() -> {
run(stop);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestConnectionImplementation.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestConnectionImplementation.java
index 89a9540..b388603 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestConnectionImplementation.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestConnectionImplementation.java
@@ -67,6 +67,7 @@ import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.apache.hadoop.hbase.util.ManualEnvironmentEdge;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
@@ -136,10 +137,10 @@ public class TestConnectionImplementation {
@Test
public void testClusterConnection() throws IOException {
- ThreadPoolExecutor otherPool = new ThreadPoolExecutor(1, 1,
- 5, TimeUnit.SECONDS,
- new SynchronousQueue<>(),
- Threads.newDaemonThreadFactory("test-hcm"));
+ ThreadPoolExecutor otherPool =
+ new ThreadPoolExecutor(1, 1, 5, TimeUnit.SECONDS, new SynchronousQueue<>(),
+ new ThreadFactoryBuilder().setNameFormat("test-hcm-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
Connection con1 = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration());
Connection con2 = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration(), otherPool);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestOpenTableInCoprocessor.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestOpenTableInCoprocessor.java
index 860a86b..8ab0ca5 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestOpenTableInCoprocessor.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestOpenTableInCoprocessor.java
@@ -43,6 +43,7 @@ import org.apache.hadoop.hbase.testclassification.CoprocessorTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -101,8 +102,9 @@ public class TestOpenTableInCoprocessor {
int maxThreads = 1;
long keepAliveTime = 60;
ThreadPoolExecutor pool =
- new ThreadPoolExecutor(1, maxThreads, keepAliveTime, TimeUnit.SECONDS,
- new SynchronousQueue<>(), Threads.newDaemonThreadFactory("hbase-table"));
+ new ThreadPoolExecutor(1, maxThreads, keepAliveTime, TimeUnit.SECONDS,
+ new SynchronousQueue<>(), new ThreadFactoryBuilder().setNameFormat("hbase-table-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
pool.allowCoreThreadTimeOut(true);
return pool;
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRegionStates.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRegionStates.java
index 6f12b95..dbad831 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRegionStates.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRegionStates.java
@@ -19,7 +19,6 @@ package org.apache.hadoop.hbase.master.assignment;
import static org.junit.Assert.assertEquals;
-import java.lang.Thread.UncaughtExceptionHandler;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorCompletionService;
@@ -36,6 +35,7 @@ import org.apache.hadoop.hbase.testclassification.MasterTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -63,13 +63,9 @@ public class TestRegionStates {
@BeforeClass
public static void setUp() throws Exception {
threadPool = Threads.getBoundedCachedThreadPool(32, 60L, TimeUnit.SECONDS,
- Threads.newDaemonThreadFactory("ProcedureDispatcher",
- new UncaughtExceptionHandler() {
- @Override
- public void uncaughtException(Thread t, Throwable e) {
- LOG.warn("Failed thread " + t.getName(), e);
- }
- }));
+ new ThreadFactoryBuilder().setNameFormat("ProcedureDispatcher-pool-%d")
+ .setUncaughtExceptionHandler((t, e) -> LOG.warn("Failed thread " + t.getName(), e))
+ .build());
executorService = new ExecutorCompletionService(threadPool);
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestMobCompactor.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestMobCompactor.java
index 7c67cd8..f14ceae 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestMobCompactor.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestMobCompactor.java
@@ -33,7 +33,6 @@ import java.util.Optional;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -93,6 +92,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -1031,18 +1031,16 @@ public class TestMobCompactor {
int maxThreads = 10;
long keepAliveTime = 60;
final SynchronousQueue<Runnable> queue = new SynchronousQueue<>();
- ThreadPoolExecutor pool = new ThreadPoolExecutor(1, maxThreads,
- keepAliveTime, TimeUnit.SECONDS, queue,
- Threads.newDaemonThreadFactory("MobFileCompactionChore"),
- new RejectedExecutionHandler() {
- @Override
- public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
- try {
- // waiting for a thread to pick up instead of throwing exceptions.
- queue.put(r);
- } catch (InterruptedException e) {
- throw new RejectedExecutionException(e);
- }
+ ThreadPoolExecutor pool =
+ new ThreadPoolExecutor(1, maxThreads, keepAliveTime, TimeUnit.SECONDS, queue,
+ new ThreadFactoryBuilder().setNameFormat("MobFileCompactionChore-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(),
+ (r, executor) -> {
+ try {
+ // waiting for a thread to pick up instead of throwing exceptions.
+ queue.put(r);
+ } catch (InterruptedException e) {
+ throw new RejectedExecutionException(e);
}
});
pool.allowCoreThreadTimeOut(true);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestPartitionedMobCompactor.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestPartitionedMobCompactor.java
index d4c70ab..0aabd31 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestPartitionedMobCompactor.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mob/compactions/TestPartitionedMobCompactor.java
@@ -33,7 +33,6 @@ import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -76,6 +75,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -904,20 +904,19 @@ public class TestPartitionedMobCompactor {
int maxThreads = 10;
long keepAliveTime = 60;
final SynchronousQueue<Runnable> queue = new SynchronousQueue<>();
- ThreadPoolExecutor pool = new ThreadPoolExecutor(1, maxThreads, keepAliveTime,
- TimeUnit.SECONDS, queue, Threads.newDaemonThreadFactory("MobFileCompactionChore"),
- new RejectedExecutionHandler() {
- @Override
- public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+ ThreadPoolExecutor pool =
+ new ThreadPoolExecutor(1, maxThreads, keepAliveTime, TimeUnit.SECONDS, queue,
+ new ThreadFactoryBuilder().setNameFormat("MobFileCompactionChore-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(),
+ (r, executor) -> {
try {
// waiting for a thread to pick up instead of throwing exceptions.
queue.put(r);
} catch (InterruptedException e) {
throw new RejectedExecutionException(e);
}
- }
- });
- ((ThreadPoolExecutor) pool).allowCoreThreadTimeOut(true);
+ });
+ pool.allowCoreThreadTimeOut(true);
return pool;
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/procedure/SimpleRSProcedureManager.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/procedure/SimpleRSProcedureManager.java
index c99fcc1..270f6d9 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/procedure/SimpleRSProcedureManager.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/procedure/SimpleRSProcedureManager.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
@@ -114,10 +115,10 @@ public class SimpleRSProcedureManager extends RegionServerProcedureManager {
}
}
- public class SimpleSubprocedurePool implements Closeable, Abortable {
+ public static class SimpleSubprocedurePool implements Closeable, Abortable {
private final ExecutorCompletionService<Void> taskPool;
- private final ThreadPoolExecutor executor;
+ private final ExecutorService executor;
private volatile boolean aborted;
private final List<Future<Void>> futures = new ArrayList<>();
private final String name;
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerReportForDuty.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerReportForDuty.java
index 17fb7db..57b4999 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerReportForDuty.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerReportForDuty.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread;
import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
import org.apache.hadoop.hbase.util.ManualEnvironmentEdge;
import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.log4j.Appender;
import org.apache.log4j.Layout;
import org.apache.log4j.PatternLayout;
@@ -232,8 +233,9 @@ public class TestRegionServerReportForDuty {
*/
@Test
public void testReportForDutyWithRSRpcRetry() throws Exception {
- ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
- new ScheduledThreadPoolExecutor(1, Threads.newDaemonThreadFactory("RSDelayedStart"));
+ ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1,
+ new ThreadFactoryBuilder().setNameFormat("RSDelayedStart-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
// Start a master and wait for it to become the active/primary master.
// Use a random unique port
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncFSWAL.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncFSWAL.java
index f31a908..bdb22ee 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncFSWAL.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncFSWAL.java
@@ -55,6 +55,7 @@ import org.apache.hadoop.hbase.wal.WALEdit;
import org.apache.hadoop.hbase.wal.WALKey;
import org.apache.hadoop.hbase.wal.WALKeyImpl;
import org.apache.hadoop.hbase.wal.WALProvider.AsyncWriter;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -82,7 +83,9 @@ public class TestAsyncFSWAL extends AbstractTestFSWAL {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
- GROUP = new NioEventLoopGroup(1, Threads.newDaemonThreadFactory("TestAsyncFSWAL"));
+ GROUP = new NioEventLoopGroup(1,
+ new ThreadFactoryBuilder().setNameFormat("TestAsyncFSWAL-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
CHANNEL_CLASS = NioSocketChannel.class;
AbstractTestFSWAL.setUpBeforeClass();
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncWALReplay.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncWALReplay.java
index 8ef3c73..e5e1f05 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncWALReplay.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestAsyncWALReplay.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hbase.testclassification.RegionServerTests;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.WAL;
import org.apache.hadoop.hbase.wal.WALFactory;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -51,7 +52,9 @@ public class TestAsyncWALReplay extends AbstractTestWALReplay {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
- GROUP = new NioEventLoopGroup(1, Threads.newDaemonThreadFactory("TestAsyncWALReplay"));
+ GROUP = new NioEventLoopGroup(1,
+ new ThreadFactoryBuilder().setNameFormat("TestAsyncWALReplay-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
CHANNEL_CLASS = NioSocketChannel.class;
Configuration conf = AbstractTestWALReplay.TEST_UTIL.getConfiguration();
conf.set(WALFactory.WAL_PROVIDER, "asyncfs");
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckMOB.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckMOB.java
index 6f175cf..12a4412 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckMOB.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckMOB.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.MiscTests;
import org.apache.hadoop.hbase.util.hbck.HFileCorruptionChecker;
import org.apache.hadoop.hbase.util.hbck.HbckTestingUtil;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -65,8 +66,10 @@ public class TestHBaseFsckMOB extends BaseTestHBaseFsck {
conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 8 * REGION_ONLINE_TIMEOUT);
TEST_UTIL.startMiniCluster(1);
- tableExecutorService = new ThreadPoolExecutor(1, POOL_SIZE, 60, TimeUnit.SECONDS,
- new SynchronousQueue<>(), Threads.newDaemonThreadFactory("testhbck"));
+ tableExecutorService =
+ new ThreadPoolExecutor(1, POOL_SIZE, 60, TimeUnit.SECONDS, new SynchronousQueue<>(),
+ new ThreadFactoryBuilder().setNameFormat("testhbck-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
hbfsckExecutorService = new ScheduledThreadPoolExecutor(POOL_SIZE);
diff --git a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/IncrementCoalescer.java b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/IncrementCoalescer.java
index 0649b26..b8707b2 100644
--- a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/IncrementCoalescer.java
+++ b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/IncrementCoalescer.java
@@ -34,6 +34,7 @@ import org.apache.hadoop.hbase.thrift.generated.TIncrement;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.metrics2.util.MBeans;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -143,9 +144,9 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
public IncrementCoalescer(ThriftHBaseServiceHandler hand) {
this.handler = hand;
LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
- pool = new ThreadPoolExecutor(CORE_POOL_SIZE, CORE_POOL_SIZE, 50,
- TimeUnit.MILLISECONDS, queue,
- Threads.newDaemonThreadFactory("IncrementCoalescer"));
+ pool = new ThreadPoolExecutor(CORE_POOL_SIZE, CORE_POOL_SIZE, 50, TimeUnit.MILLISECONDS, queue,
+ new ThreadFactoryBuilder().setNameFormat("IncrementCoalescer-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
MBeans.register("thrift", "Thrift", this);
}
diff --git a/hbase-zookeeper/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKWatcher.java b/hbase-zookeeper/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKWatcher.java
index fca8738..777776f 100644
--- a/hbase-zookeeper/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKWatcher.java
+++ b/hbase-zookeeper/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKWatcher.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.hbase.security.Superusers;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
@@ -95,8 +96,9 @@ public class ZKWatcher implements Watcher, Abortable, Closeable {
// and further prevents deadlocks if the process method itself makes other zookeeper calls.
// It is ok to do it in a single thread because the Zookeeper ClientCnxn already serializes the
// requests using a single while loop and hence there is no performance degradation.
- private final ExecutorService zkEventProcessor =
- Executors.newSingleThreadExecutor(Threads.getNamedThreadFactory("zk-event-processor"));
+ private final ExecutorService zkEventProcessor = Executors.newSingleThreadExecutor(
+ new ThreadFactoryBuilder().setNameFormat("zk-event-processor-pool-%d")
+ .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
private final Configuration conf;