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

ForkingTaskRunner: Upload task logs even when job fails

This commit is contained in:
Gian Merlino 2014-01-08 14:46:18 -08:00
parent 7f430d9fde
commit 2c53af4d66
1 changed files with 14 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
indexing-service/src/main/java/io/druid/indexing/overlord/ForkingTaskRunner.java
View File

@ -218,29 +218,20 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogStreamer
} }
log.info("Logging task %s output to: %s", task.getId(), logFile); log.info("Logging task %s output to: %s", task.getId(), logFile);
final InputStream fromProc = processHolder.process.getInputStream();
final OutputStream toLogfile = closer.register(
Files.newOutputStreamSupplier(logFile).getOutput()
);
boolean runFailed = true; boolean runFailed = true;
ByteStreams.copy(fromProc, toLogfile); try (final OutputStream toLogfile = Files.newOutputStreamSupplier(logFile).getOutput()) {
final int statusCode = processHolder.process.waitFor(); ByteStreams.copy(processHolder.process.getInputStream(), toLogfile);
log.info("Process exited with status[%d] for task: %s", statusCode, task.getId()); final int statusCode = processHolder.process.waitFor();
log.info("Process exited with status[%d] for task: %s", statusCode, task.getId());
if (statusCode == 0) { if (statusCode == 0) {
runFailed = false; runFailed = false;
}
}
finally {
// Upload task logs
taskLogPusher.pushTaskLog(task.getId(), logFile);
} }
// Upload task logs
// XXX: Consider uploading periodically for very long-lived tasks to prevent
// XXX: bottlenecks at the end or the possibility of losing a lot of logs all
// XXX: at once.
taskLogPusher.pushTaskLog(task.getId(), logFile);
if (!runFailed) { if (!runFailed) {
// Process exited successfully // Process exited successfully
@ -255,9 +246,9 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogStreamer
closer.close(); closer.close();
} }
} }
catch (Exception e) { catch (Throwable t) {
log.info(e, "Exception caught during execution"); log.info(t, "Exception caught during execution");
throw Throwables.propagate(e); throw Throwables.propagate(t);
} }
finally { finally {
try { try {