Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merger: Tweaks to task shutdown

This commit is contained in:
Gian Merlino 2013-04-15 18:51:01 -07:00
parent d879c9a41e
commit cdbdf843c0
1 changed files with 9 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
merger/src/main/java/com/metamx/druid/merger/coordinator/ForkingTaskRunner.java
View File

@ -125,11 +125,12 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogProvider
// time to adjust process holders // time to adjust process holders
synchronized (tasks) { synchronized (tasks) {
if (Thread.interrupted()) { final TaskInfo taskInfo = tasks.get(task.getId());
throw new InterruptedException();
if (taskInfo.shutdown) {
throw new IllegalStateException("Task has been shut down!");
} }
final TaskInfo taskInfo = tasks.get(task.getId());
if (taskInfo == null) { if (taskInfo == null) {
throw new ISE("WTF?! TaskInfo disappeared for task: %s", task.getId()); throw new ISE("WTF?! TaskInfo disappeared for task: %s", task.getId());
} }
@ -227,11 +228,8 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogProvider
return TaskStatus.failure(task.getId()); return TaskStatus.failure(task.getId());
} }
} }
catch (InterruptedException e) { catch (Exception e) {
log.info(e, "Interrupted during execution"); log.info(e, "Exception caught during execution");
return TaskStatus.failure(task.getId());
}
catch (IOException e) {
throw Throwables.propagate(e); throw Throwables.propagate(e);
} }
finally { finally {
@ -288,9 +286,9 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogProvider
log.info("Ignoring request to cancel unknown task: %s", taskid); log.info("Ignoring request to cancel unknown task: %s", taskid);
return; return;
} }
}
taskInfo.statusFuture.cancel(true); taskInfo.shutdown = true;
}
if (taskInfo.processHolder != null) { if (taskInfo.processHolder != null) {
final int shutdowns = taskInfo.processHolder.shutdowns.getAndIncrement(); final int shutdowns = taskInfo.processHolder.shutdowns.getAndIncrement();
@ -397,6 +395,7 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogProvider
private static class TaskInfo private static class TaskInfo
{ {
private final ListenableFuture<TaskStatus> statusFuture; private final ListenableFuture<TaskStatus> statusFuture;
private volatile boolean shutdown = false;
private volatile ProcessHolder processHolder = null; private volatile ProcessHolder processHolder = null;
private TaskInfo(ListenableFuture<TaskStatus> statusFuture) private TaskInfo(ListenableFuture<TaskStatus> statusFuture)