fix task executor logging
Closes from kadaan/GOBBLIN-
359__Fix_TaskExecutor_Logging
@@ -136,16 +136,18 @@ public class TaskExecutor extends AbstractIdleService {
Preconditions.checkArgument(queuedTaskTimeMaxAge > 0, "Queued task time max age should be positive");
// Currently a fixed-size thread pool is used to execute tasks. We probably need to revisit this later.
- this.taskExecutor = Executors.newScheduledThreadPool(
+ this.taskExecutor = ExecutorsUtils.loggingDecorator(Executors.newScheduledThreadPool(
taskExecutorThreadPoolSize,
- ExecutorsUtils.newThreadFactory(Optional.of(LOG), Optional.of("TaskExecutor-%d")));
+ ExecutorsUtils.newThreadFactory(Optional.of(LOG), Optional.of("TaskExecutor-%d"))));
this.retryIntervalInSeconds = retryIntervalInSeconds;
this.queuedTaskTimeMaxSize = queuedTaskTimeMaxSize;
this.queuedTaskTimeMaxAge = queuedTaskTimeMaxAge;
- this.forkExecutor = new ThreadPoolExecutor(
- // The core thread pool size is equal to that of the task executor as there's at least one fork per task
+ this.forkExecutor = ExecutorsUtils.loggingDecorator(
+ new ThreadPoolExecutor(
+ // The core thread pool size is equal to that of the task
+ // executor as there's at least one fork per task
taskExecutorThreadPoolSize,
// The fork executor thread pool size is essentially unbounded. This is to make sure all forks of
// a task get a thread to run so all forks of the task are making progress. This is necessary since
@@ -157,7 +159,7 @@ public class TaskExecutor extends AbstractIdleService {
TimeUnit.MILLISECONDS,
// The work queue is a SynchronousQueue. This essentially forces a new thread to be created for each fork.
new SynchronousQueue<Runnable>(),
- ExecutorsUtils.newThreadFactory(Optional.of(LOG), Optional.of("ForkExecutor-%d")));
+ ExecutorsUtils.newThreadFactory(Optional.of(LOG), Optional.of("ForkExecutor-%d"))));
}
/**
