From c21d7f6ee2cab8d49cabd7582b40885d19d0a5d2 Mon Sep 17 00:00:00 2001
From: Gian Merlino <gianmerlino@gmail.com>
Date: Fri, 12 Apr 2013 14:06:40 -0700
Subject: [PATCH] ForkingTaskRunner: Slight tweaks to setup and teardown

---
 .../merger/coordinator/ForkingTaskRunner.java | 29 +++++++++----------
 1 file changed, 13 insertions(+), 16 deletions(-)

diff --git a/merger/src/main/java/com/metamx/druid/merger/coordinator/ForkingTaskRunner.java b/merger/src/main/java/com/metamx/druid/merger/coordinator/ForkingTaskRunner.java
index 1c6099aa083..b6ebdf4276f 100644
--- a/merger/src/main/java/com/metamx/druid/merger/coordinator/ForkingTaskRunner.java
+++ b/merger/src/main/java/com/metamx/druid/merger/coordinator/ForkingTaskRunner.java
@@ -142,14 +142,9 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogProvider
                             final int childPort = findUnusedPort();
                             final String childHost = String.format(config.getHostPattern(), childPort);
 
-                            Iterables.addAll(
-                                command,
-                                ImmutableList.of(
-                                    config.getJavaCommand(),
-                                    "-cp",
-                                    config.getJavaClasspath()
-                                )
-                            );
+                            command.add(config.getJavaCommand());
+                            command.add("-cp");
+                            command.add(config.getJavaClasspath());
 
                             Iterables.addAll(
                                 command,
@@ -195,25 +190,27 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogProvider
                           final InputStream fromProc = processHolder.process.getInputStream();
                           final OutputStream toLogfile = Files.newOutputStreamSupplier(logFile).getOutput();
 
-                          boolean copyFailed = false;
+                          boolean runFailed = false;
 
                           try {
                             ByteStreams.copy(fromProc, toLogfile);
+                            final int statusCode = processHolder.process.waitFor();
+                            log.info("Process exited with status[%d] for task: %s", statusCode, task.getId());
+
+                            if (statusCode != 0) {
+                              runFailed = true;
+                            }
                           }
                           catch (Exception e) {
                             log.warn(e, "Failed to read from process for task: %s", task.getId());
-                            copyFailed = true;
+                            runFailed = true;
                           }
                           finally {
                             Closeables.closeQuietly(fromProc);
                             Closeables.closeQuietly(toLogfile);
+                            Closeables.closeQuietly(toProc);
                           }
 
-                          final int statusCode = processHolder.process.waitFor();
-
-                          log.info("Process exited with status[%d] for task: %s", statusCode, task.getId());
-                          Closeables.closeQuietly(toProc);
-
                           // Upload task logs
 
                           // XXX: Consider uploading periodically for very long-lived tasks to prevent
@@ -222,7 +219,7 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogProvider
 
                           taskLogPusher.pushTaskLog(task.getId(), logFile);
 
-                          if (!copyFailed && statusCode == 0) {
+                          if (!runFailed) {
                             // Process exited successfully
                             return jsonMapper.readValue(statusFile, TaskStatus.class);
                           } else {