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/d999a3b2 Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/d999a3b2 Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/d999a3b2 Branch: refs/heads/branch-3 Commit: d999a3b23c1306712f91338efa8757b845114954 Parents: ca0d4f7 Author: Gopal V <[email protected]> Authored: Tue Jul 31 13:51:53 2018 -0700 Committer: Gopal V <[email protected]> Committed: Tue Jul 31 13:53:22 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/d999a3b2/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 78980fa..9bc4d2a 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 @@ -978,21 +978,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 {
