presto-main/src/main/java/io/prestosql/operator/WorkProcessorUtils.java/WorkProcessorUtils/processStateMonitor
class WorkProcessorUtils:    static <T> WorkProcessor<T> processStateMonitor(WorkProcessor<T> processor, Consumer<ProcessState<T>> monitor)
    {
        requireNonNull(processor, "processor is null");
        requireNonNull(monitor, "monitor is null");
        return WorkProcessor.create(() -> {
            ProcessState<T> state = getNextState(processor);
            monitor.accept(state);
            return state;
        });
    }