hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From w...@apache.org
Subject hive git commit: HIVE-13833 : Add an initial delay when starting the heartbeat (Wei Zheng, reviewed by Eugene Koifman)
Date Tue, 14 Jun 2016 22:31:04 GMT
Repository: hive
Updated Branches:
  refs/heads/master 942f731c7 -> 40a15536a


HIVE-13833 : Add an initial delay when starting the heartbeat (Wei Zheng, 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/40a15536
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/40a15536
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/40a15536

Branch: refs/heads/master
Commit: 40a15536a37893940a4ecfc06c8eda46cd5d50ea
Parents: 942f731
Author: Wei Zheng <weiz@apache.org>
Authored: Tue Jun 14 15:30:56 2016 -0700
Committer: Wei Zheng <weiz@apache.org>
Committed: Tue Jun 14 15:30:56 2016 -0700

----------------------------------------------------------------------
 .../hadoop/hive/ql/lockmgr/DbTxnManager.java    | 45 ++++++++++----------
 1 file changed, 22 insertions(+), 23 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/40a15536/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 9988eec..5b6f20c 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
@@ -20,7 +20,6 @@ package org.apache.hadoop.hive.ql.lockmgr;
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.ql.io.AcidUtils;
-import org.apache.hadoop.hive.ql.parse.SemanticAnalyzer;
 import org.apache.hive.common.util.ShutdownHookManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -84,7 +83,7 @@ public class DbTxnManager extends HiveTxnManagerImpl {
   private static ScheduledExecutorService heartbeatExecutorService = null;
   private ScheduledFuture<?> heartbeatTask = null;
   private Runnable shutdownRunner = null;
-  static final int SHUTDOWN_HOOK_PRIORITY = 0;
+  private static final int SHUTDOWN_HOOK_PRIORITY = 0;
 
   DbTxnManager() {
     shutdownRunner = new Runnable() {
@@ -161,10 +160,11 @@ public class DbTxnManager extends HiveTxnManagerImpl {
     getLockManager();
 
     boolean atLeastOneLock = false;
+    queryId = plan.getQueryId();
 
-    LockRequestBuilder rqstBuilder = new LockRequestBuilder(plan.getQueryId());
+    LockRequestBuilder rqstBuilder = new LockRequestBuilder(queryId);
     //link queryId to txnId
-    LOG.info("Setting lock request transaction to " + JavaUtils.txnIdToString(txnId) + "
for queryId=" + plan.getQueryId());
+    LOG.info("Setting lock request transaction to " + JavaUtils.txnIdToString(txnId) + "
for queryId=" + queryId);
     rqstBuilder.setTransactionId(txnId)
         .setUser(username);
 
@@ -304,7 +304,7 @@ public class DbTxnManager extends HiveTxnManagerImpl {
     // Make sure we need locks.  It's possible there's nothing to lock in
     // this operation.
     if (!atLeastOneLock) {
-      LOG.debug("No locks needed for queryId" + plan.getQueryId());
+      LOG.debug("No locks needed for queryId" + queryId);
       return null;
     }
 
@@ -312,7 +312,7 @@ public class DbTxnManager extends HiveTxnManagerImpl {
     if(isTxnOpen()) {
       statementId++;
     }
-    LockState lockState = lockMgr.lock(rqstBuilder.build(), plan.getQueryId(), isBlocking,
locks);
+    LockState lockState = lockMgr.lock(rqstBuilder.build(), queryId, isBlocking, locks);
     ctx.setHiveLocks(locks);
     return lockState;
   }
@@ -324,15 +324,13 @@ public class DbTxnManager extends HiveTxnManagerImpl {
     return t;
   }
   /**
-   * This is for testing only.
+   * This is for testing only. Normally client should call {@link #acquireLocks(QueryPlan,
Context, String, boolean)}
    * @param delay time to delay for first heartbeat
-   * @return null if no locks were needed
    */
   @VisibleForTesting
   void acquireLocksWithHeartbeatDelay(QueryPlan plan, Context ctx, String username, long
delay) throws LockException {
     acquireLocks(plan, ctx, username, true);
     ctx.setHeartbeater(startHeartbeat(delay));
-    queryId = plan.getQueryId();
   }
 
 
@@ -439,24 +437,25 @@ public class DbTxnManager extends HiveTxnManagerImpl {
     }
   }
 
-  private Heartbeater startHeartbeat() throws LockException {
-    return startHeartbeat(0);
-  }
-
   /**
-   *  This is for testing only.  Normally client should call {@link #startHeartbeat()}
-   *  Make the heartbeater start before an initial delay period.
-   *  @param delay time to delay before first execution, in milliseconds
-   *  @return heartbeater
+   * Start the heartbeater threadpool and return the task.
+   * @param initialDelay time to delay before first execution, in milliseconds
+   * @return heartbeater
    */
-  Heartbeater startHeartbeat(long delay) throws LockException {
+  private Heartbeater startHeartbeat(long initialDelay) throws LockException {
     long heartbeatInterval = getHeartbeatInterval(conf);
     assert heartbeatInterval > 0;
     Heartbeater heartbeater = new Heartbeater(this, conf);
+    // For negative testing purpose..
+    if(conf.getBoolVar(HiveConf.ConfVars.HIVE_IN_TEST) && conf.getBoolVar(HiveConf.ConfVars.HIVETESTMODEFAILHEARTBEATER))
{
+      initialDelay = 0;
+    } else if (initialDelay == 0) {
+      initialDelay = heartbeatInterval;
+    }
     heartbeatTask = heartbeatExecutorService.scheduleAtFixedRate(
-        heartbeater, delay, heartbeatInterval, TimeUnit.MILLISECONDS);
-    LOG.info("Started heartbeat with delay/interval = " + 0 + "/" + heartbeatInterval + "
" +
-        TimeUnit.MILLISECONDS + " for query: " + queryId);
+        heartbeater, initialDelay, heartbeatInterval, TimeUnit.MILLISECONDS);
+    LOG.info("Started heartbeat with delay/interval = " + initialDelay + "/" + heartbeatInterval
+
+        " " + TimeUnit.MILLISECONDS + " for query: " + queryId);
     return heartbeater;
   }
 
@@ -584,7 +583,7 @@ public class DbTxnManager extends HiveTxnManagerImpl {
     return statementId;
   }
 
-  public static long getHeartbeatInterval(Configuration conf) throws LockException {
+  private static long getHeartbeatInterval(Configuration conf) throws LockException {
     // Retrieve HIVE_TXN_TIMEOUT in MILLISECONDS (it's defined as SECONDS),
     // then divide it by 2 to give us a safety factor.
     long interval =
@@ -612,7 +611,7 @@ public class DbTxnManager extends HiveTxnManagerImpl {
      *
      * @param txnMgr transaction manager for this operation
      */
-    public Heartbeater(HiveTxnManager txnMgr, HiveConf conf) {
+    Heartbeater(HiveTxnManager txnMgr, HiveConf conf) {
       this.txnMgr = txnMgr;
       this.conf = conf;
       lockException = null;


Mime
View raw message