kgyrtkirk commented on a change in pull request #794: HIVE-21884
URL: https://github.com/apache/hive/pull/794#discussion_r335690222
##########
File path:
ql/src/java/org/apache/hadoop/hive/ql/schq/ScheduledQueryExecutionService.java
##########
@@ -0,0 +1,159 @@
+package org.apache.hadoop.hive.ql.schq;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.QueryState;
+import org.apache.hadoop.hive.metastore.api.ScheduledQueryPollResponse;
+import org.apache.hadoop.hive.metastore.api.ScheduledQueryProgressInfo;
+import org.apache.hadoop.hive.ql.DriverFactory;
+import org.apache.hadoop.hive.ql.IDriver;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+public class ScheduledQueryExecutionService implements Closeable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ScheduledQueryExecutionService.class);
+
+ private ScheduledQueryExecutionContext context;
+ private ScheduledQueryExecutor worker;
+
+ public static ScheduledQueryExecutionService
startScheduledQueryExecutorService(HiveConf conf) {
+ MetastoreBasedScheduledQueryService qService = new
MetastoreBasedScheduledQueryService(conf);
+ ExecutorService executor =
+ Executors.newCachedThreadPool(
+ new
ThreadFactoryBuilder().setDaemon(true).setNameFormat("Scheduled Query Thread
%d").build());
+ ScheduledQueryExecutionContext ctx = new
ScheduledQueryExecutionContext(executor, conf, qService);
+ return new ScheduledQueryExecutionService(ctx);
+ }
+
+ public ScheduledQueryExecutionService(ScheduledQueryExecutionContext ctx) {
+ context = ctx;
+ ctx.executor.submit(worker = new ScheduledQueryExecutor());
+ ctx.executor.submit(new ProgressReporter());
+ }
+
+ class ScheduledQueryExecutor implements Runnable {
+
+ private ScheduledQueryProgressInfo info;
+
+ @Override
+ public void run() {
+ while (true) {
+ ScheduledQueryPollResponse q =
context.schedulerService.scheduledQueryPoll();
+ if (q.isSetExecutionId()) {
+ try{
+ processQuery(q);
+ } catch (Throwable t) {
+ LOG.error("Unexpected exception during scheduled query
processing", t);
+ }
+ } else {
+ try {
+ Thread.sleep(context.getIdleSleepTime());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.info("interrupted");
+ break;
+ }
+ }
+ }
+ }
+
+ QueryState lastState;
+ public synchronized void reportQueryProgress() {
Review comment:
the reason of the synch here is to lock out problematic updates to the
actually executing scheduled query.
* the scheduled query is about to finish; and it sends the "FINISHED" state
* meanwhile the periodic reporting kicks in and wants to send that the state
is "EXECUTING"
the synch is here to lock out the latter while the state change happens ;
and then it will realise that no update is needed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
