You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lens.apache.org by pr...@apache.org on 2016/06/08 13:35:49 UTC
lens git commit: LENS-1169: Improve the logic of stopping query
service
Repository: lens
Updated Branches:
refs/heads/master 3e7a6f602 -> 9ace50b3e
LENS-1169: Improve the logic of stopping query service
Project: http://git-wip-us.apache.org/repos/asf/lens/repo
Commit: http://git-wip-us.apache.org/repos/asf/lens/commit/9ace50b3
Tree: http://git-wip-us.apache.org/repos/asf/lens/tree/9ace50b3
Diff: http://git-wip-us.apache.org/repos/asf/lens/diff/9ace50b3
Branch: refs/heads/master
Commit: 9ace50b3e657922de0430df5e1a44d0571ebd699
Parents: 3e7a6f6
Author: Rajat Khandelwal <pr...@apache.org>
Authored: Wed Jun 8 19:05:05 2016 +0530
Committer: Rajat Khandelwal <ra...@gmail.com>
Committed: Wed Jun 8 19:05:05 2016 +0530
----------------------------------------------------------------------
.../server/query/QueryExecutionServiceImpl.java | 72 ++++++++++++++------
.../lens/server/query/QueryResultPurger.java | 10 ++-
.../server/query/TestQueryResultPurger.java | 2 +-
3 files changed, 59 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lens/blob/9ace50b3/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java b/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
index 8fe02aa..23c7743 100644
--- a/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
+++ b/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
@@ -1223,6 +1223,22 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
}
+ private void awaitTermination(ExecutorService service) {
+ try {
+ service.awaitTermination(1, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ log.info("Couldn't finish executor service within 1 minute: {}", service);
+ }
+ }
+
+ private void awaitTermination(QueryResultPurger service) {
+ try {
+ service.awaitTermination(1, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ log.info("Couldn't finish query result purger within 1 minute: {}", service);
+ }
+ }
+
/*
* (non-Javadoc)
*
@@ -1230,23 +1246,29 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
*/
public void prepareStopping() {
super.prepareStopping();
- querySubmitter.interrupt();
- statusPoller.interrupt();
- queryPurger.interrupt();
- prepareQueryPurger.interrupt();
- }
+ Thread[] threadsToStop = new Thread[]{querySubmitter, statusPoller, queryPurger, prepareQueryPurger};
+ // Nudge the threads to stop
+ for (Thread th : threadsToStop) {
+ th.interrupt();
+ }
- /*
- * (non-Javadoc)
- *
- * @see org.apache.hive.service.CompositeService#stop()
- */
- public synchronized void stop() {
- super.stop();
+ // Nudge executor pools to stop
- waitingQueriesSelectionSvc.shutdown();
+ // Hard shutdown, since it doesn't matter whether waiting queries were selected, all will be
+ // selected in the next restart
+ waitingQueriesSelectionSvc.shutdownNow();
+ // Soft shutdown, Wait for current estimate tasks
+ estimatePool.shutdown();
+ // Soft shutdown for result purger too. Purging shouldn't take much time.
+ if (null != queryResultPurger) {
+ queryResultPurger.shutdown();
+ }
+ // Soft shutdown right now, will await termination in this method itself, since cancellation pool
+ // should be terminated before query state gets persisted.
+ queryCancellationPool.shutdown();
- for (Thread th : new Thread[]{querySubmitter, statusPoller, queryPurger, prepareQueryPurger}) {
+ // Join the threads.
+ for (Thread th : threadsToStop) {
try {
log.debug("Waiting for {}", th.getName());
th.join();
@@ -1254,15 +1276,21 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
log.error("Error waiting for thread: {}", th.getName(), e);
}
}
+ // Needs to be done before queries' states are persisted, hence doing here. Await of other
+ // executor services can be done after persistence, hence they are done in #stop
+ awaitTermination(queryCancellationPool);
+ }
- estimatePool.shutdownNow();
-
- if (null != queryResultPurger) {
- queryResultPurger.stop();
- }
-
- queryCancellationPool.shutdown();
-
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.hive.service.CompositeService#stop()
+ */
+ public synchronized void stop() {
+ super.stop();
+ awaitTermination(waitingQueriesSelectionSvc);
+ awaitTermination(estimatePool);
+ awaitTermination(queryResultPurger);
log.info("Query execution service stopped");
}
http://git-wip-us.apache.org/repos/asf/lens/blob/9ace50b3/lens-server/src/main/java/org/apache/lens/server/query/QueryResultPurger.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/query/QueryResultPurger.java b/lens-server/src/main/java/org/apache/lens/server/query/QueryResultPurger.java
index 2be11ea..756c282 100644
--- a/lens-server/src/main/java/org/apache/lens/server/query/QueryResultPurger.java
+++ b/lens-server/src/main/java/org/apache/lens/server/query/QueryResultPurger.java
@@ -145,9 +145,9 @@ public class QueryResultPurger implements Runnable {
/**
* Stops query result purger
*/
- public void stop() {
+ public void shutdown() {
if (null != queryResultPurgerExecutor) {
- queryResultPurgerExecutor.shutdownNow();
+ queryResultPurgerExecutor.shutdown();
log.info("Stopped query result purger.");
}
}
@@ -174,4 +174,10 @@ public class QueryResultPurger implements Runnable {
}
return metricsService;
}
+
+ public void awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+ if (null != queryResultPurgerExecutor) {
+ queryResultPurgerExecutor.awaitTermination(timeout, unit);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/9ace50b3/lens-server/src/test/java/org/apache/lens/server/query/TestQueryResultPurger.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/query/TestQueryResultPurger.java b/lens-server/src/test/java/org/apache/lens/server/query/TestQueryResultPurger.java
index 9aeb645..c498ada 100644
--- a/lens-server/src/test/java/org/apache/lens/server/query/TestQueryResultPurger.java
+++ b/lens-server/src/test/java/org/apache/lens/server/query/TestQueryResultPurger.java
@@ -74,7 +74,7 @@ public class TestQueryResultPurger {
QueryResultPurger queryResultPurger = new QueryResultPurger();
queryResultPurger.init(conf);
Thread.sleep(2000); // sleep for 2 seconds, enough to run query purger
- queryResultPurger.stop();
+ queryResultPurger.shutdown();
verify(conf.get(LensConfConstants.RESULT_SET_PARENT_DIR), 1);
verify(conf.get(LensConfConstants.RESULT_SET_PARENT_DIR) + "/" + conf.get(LensConfConstants.QUERY_HDFS_OUTPUT_PATH),
0);