presto-main/src/main/java/io/prestosql/operator/WorkProcessorPipelineSourceOperator.java/WorkProcessorPipelineSourceOperator/closeOperators
class WorkProcessorPipelineSourceOperator:    private void closeOperators(int lastOperatorIndex)
    {
        // record the current interrupted status (and clear the flag); we'll reset it later
        boolean wasInterrupted = Thread.interrupted();
        Throwable inFlightException = null;
        try {
            for (int i = 0; i <= lastOperatorIndex; ++i) {
                WorkProcessorOperatorContext workProcessorOperatorContext = workProcessorOperatorContexts.get(i);
                WorkProcessorOperator operator = workProcessorOperatorContext.operator;
                if (operator == null) {
                    // operator is already closed
                    continue;
                }

                try {
                    operator.close();
                }
                catch (InterruptedException t) {
                    // don't record the stack
                    wasInterrupted = true;
                }
                catch (Throwable t) {
                    inFlightException = handleOperatorCloseError(
                            inFlightException,
                            t,
                            "Error closing WorkProcessor operator %s for task %s",
                            workProcessorOperatorContext.operatorId,
                            operatorContext.getDriverContext().getTaskId());
                }
                finally {
                    workProcessorOperatorContext.memoryTrackingContext.close();
                    workProcessorOperatorContext.finalOperatorInfo = operator.getOperatorInfo().orElse(null);
                    workProcessorOperatorContext.operator = null;
                }
            }
        }
        finally {
            // reset the interrupted flag
            if (wasInterrupted) {
                Thread.currentThread().interrupt();
            }
        }
        if (inFlightException != null) {
            throwIfUnchecked(inFlightException);
            throw new RuntimeException(inFlightException);
        }
    }