Remove inner StateChangeListener class
@@ -241,9 +241,9 @@ public class SqlTaskExecution
// don't register the task if it is already completed (most likely failed during planning above)
if (!taskStateMachine.getState().isDone()) {
taskHandle = taskExecutor.addTask(taskId, outputBuffer::getUtilization, getInitialSplitsPerNode(taskContext.getSession()), getSplitConcurrencyAdjustmentInterval(taskContext.getSession()));
- taskStateMachine.addStateChangeListener(new RemoveTaskHandleWhenDone(taskExecutor, taskHandle));
taskStateMachine.addStateChangeListener(state -> {
if (state.isDone()) {
+ taskExecutor.removeTask(taskHandle);
for (DriverFactory factory : localExecutionPlan.getDriverFactories()) {
factory.noMoreDrivers();
}
@@ -996,27 +996,6 @@ public class SqlTaskExecution
}
}
- private static final class RemoveTaskHandleWhenDone
- implements StateChangeListener<TaskState>
- {
- private final TaskExecutor taskExecutor;
- private final TaskHandle taskHandle;
-
- private RemoveTaskHandleWhenDone(TaskExecutor taskExecutor, TaskHandle taskHandle)
- {
- this.taskExecutor = requireNonNull(taskExecutor, "taskExecutor is null");
- this.taskHandle = requireNonNull(taskHandle, "taskHandle is null");
- }
-
- @Override
- public void stateChanged(TaskState newState)
- {
- if (newState.isDone()) {
- taskExecutor.removeTask(taskHandle);
- }
- }
- }
-
private static final class CheckTaskCompletionOnBufferFinish
implements StateChangeListener<BufferState>
{
