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 9f935f07bc47..039ff4d12f8a 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.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 ccdfec7f4f6f..a7e715c36830 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.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 @@ public boolean isDeadServer(ServerName sn) { @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 06e243bf4c45..b3b7b7db486c 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.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 @@ private ThreadPoolExecutor getThreadPool(int maxThreads, int coreThreads, String 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 51e45b17fa23..a6866d203f0b 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.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.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 static ThreadPoolExecutor getDefaultExecutor(Configuration conf) { // 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 bf2f36157359..064664e64d8a 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.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 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 7c9d9275e318..3fd6ac9b5e15 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.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 f750fd656102..f4a27feed501 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.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 void beforeEach() { 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 41504843e5c7..f1090f5b2b5a 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.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 int run(String[] arg0) throws Exception { // 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 352734064ab3..ab6f805fc825 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 @@ @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 static void sleepWithoutInterrupt(final long msToWait) { * @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 static void setLoggingUncaughtExceptionHandler(Thread t) { 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 b8b321395cf6..8e34af9b6ad8 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.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 int run(String[] args) throws Exception { 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 ecb4ffcc0d8e..dd76083da938 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.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 boolean start() { // 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 825746695f2c..608f118b5593 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.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 interface Publisher extends Closeable { @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 c9cc906f16c1..2d91c5529757 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.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 @@ private Pair splitStoreFiles(final MasterProcedureEnv env, 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>> futures = new ArrayList>>(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 2a6e6376ed9a..a9578881ca1a 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.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.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 static ExecutorService createMobCompactorThreadPool(Configuration conf) { } final SynchronousQueue 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 cb3512ae94b2..b5e743eb2eb9 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.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 @@ private NamedQueueRecorder(Configuration conf) { // 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 1537f7c9ec6a..744bbbef7971 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.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 @@ void interruptIfNecessary() { } 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 8fb8e83e00f4..7776f3dc0a59 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 FSHLog(final FileSystem fs, final Path rootDir, final String logDir, // 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 cead808e1243..7eca732d5837 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.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 @@ private ExecutorService getDefaultThreadPool(Configuration conf) { 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 6a2158104946..b6eb7a7d1e1b 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.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.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 @@ private ThreadPoolExecutor createExecutor(final String name) { 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 3120d8adadf4..53f03029cad3 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.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 @@ private ExecutorService createThreadPool(Configuration conf) { 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 f19f864f3b64..3b22de46adc0 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.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 boolean accept(Path path) { // 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 1783d7d7a38b..a58c681821b5 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.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 HBaseFsck(Configuration conf) throws IOException, ClassNotFoundException 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 c0189c83aae2..99691c8ac3af 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.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 @@ static ThreadPoolExecutor getRegionOpenAndInitThreadPool(final Configuration con "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 58057932bd73..58bd4ddc683b 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 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 @@ private void createBatchPool(Configuration conf) { * 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 60a190ace217..826febeaaefa 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.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 OutputSink(WALSplitter.PipelineController controller, EntryBuffers entryB 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 45648062fd78..a58648f45bbc 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.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 @@ private ExecutorService createThreadPool() { BlockingQueue workQueue = new LinkedBlockingQueue( 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 94adce72bb0e..6d89299a7e52 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.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 void test() throws Exception { 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> 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 89a954020f75..b388603bcfab 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.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 void tearDown() throws IOException { @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 860a86b1e5ae..8ab0ca578601 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.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 @@ private ExecutorService getPool() { 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 6f12b95510eb..dbad831dcfe4 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 @@ 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.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 7c67cd8e79c1..f14ceaede04d 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.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.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 @@ private static ExecutorService createThreadPool(Configuration conf) { int maxThreads = 10; long keepAliveTime = 60; final SynchronousQueue 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 d4c70ab85f01..0aabd3114e4f 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.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.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 @@ private static ExecutorService createThreadPool() { int maxThreads = 10; long keepAliveTime = 60; final SynchronousQueue 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 c99fcc19a2f0..270f6d95c9e2 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.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 Subprocedure buildSubprocedure(String name, byte[] data) { } } - public class SimpleSubprocedurePool implements Closeable, Abortable { + public static class SimpleSubprocedurePool implements Closeable, Abortable { private final ExecutorCompletionService taskPool; - private final ThreadPoolExecutor executor; + private final ExecutorService executor; private volatile boolean aborted; private final List> 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 17fb7db06543..57b499926061 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.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 void testReportForDutyWithMasterChange() throws Exception { */ @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 f31a908000b1..bdb22ee6ec00 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.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 8ef3c73e47bb..e5e1f0540e84 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.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 6f175cf90a8a..12a4412f1542 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.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 static void setUpBeforeClass() throws Exception { 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 0649b26e25e6..b8707b20cdcc 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.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 boolean equals(Object obj) { public IncrementCoalescer(ThriftHBaseServiceHandler hand) { this.handler = hand; LinkedBlockingQueue 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 fca8738778a4..777776ffc725 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.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;