HIVE-19199: ACID: DbTxnManager heartbeat-service needs static sync init (Gopal V, reviewed by Eugene Koifman)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/3e46515d Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/3e46515d Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/3e46515d Branch: refs/heads/master Commit: 3e46515d3d556205cb643908565a24f6634c27c1 Parents: 80ae333 Author: Gopal V <[email protected]> Authored: Tue Jul 31 13:51:53 2018 -0700 Committer: Gopal V <[email protected]> Committed: Tue Jul 31 13:51:53 2018 -0700 ---------------------------------------------------------------------- .../hadoop/hive/ql/lockmgr/DbTxnManager.java | 31 +++++++++++--------- 1 file changed, 17 insertions(+), 14 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/3e46515d/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java b/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java index 27abaf5..f746d2d 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java @@ -785,21 +785,24 @@ public final class DbTxnManager extends HiveTxnManagerImpl { } private synchronized void initHeartbeatExecutorService() { - if (heartbeatExecutorService != null && !heartbeatExecutorService.isShutdown() - && !heartbeatExecutorService.isTerminated()) { - return; + synchronized (DbTxnManager.class) { + if (heartbeatExecutorService != null && !heartbeatExecutorService.isShutdown() + && !heartbeatExecutorService.isTerminated()) { + return; + } + heartbeatExecutorService = + Executors.newScheduledThreadPool( + conf.getIntVar(HiveConf.ConfVars.HIVE_TXN_HEARTBEAT_THREADPOOL_SIZE), + new ThreadFactory() { + private final AtomicInteger threadCounter = new AtomicInteger(); + + @Override + public Thread newThread(Runnable r) { + return new HeartbeaterThread(r, "Heartbeater-" + threadCounter.getAndIncrement()); + } + }); + ((ScheduledThreadPoolExecutor) heartbeatExecutorService).setRemoveOnCancelPolicy(true); } - heartbeatExecutorService = - Executors.newScheduledThreadPool( - conf.getIntVar(HiveConf.ConfVars.HIVE_TXN_HEARTBEAT_THREADPOOL_SIZE), new ThreadFactory() { - private final AtomicInteger threadCounter = new AtomicInteger(); - - @Override - public Thread newThread(Runnable r) { - return new HeartbeaterThread(r, "Heartbeater-" + threadCounter.getAndIncrement()); - } - }); - ((ScheduledThreadPoolExecutor) heartbeatExecutorService).setRemoveOnCancelPolicy(true); } public static class HeartbeaterThread extends Thread {
