`ExchangeService$InactiveSinksReaper` runs its task on the scheduler thread. It's not clear to me if it's lightweight enough for that to be the right thing to do or not. If it is lightweight enough that it can run here, could we add a comment to that effect somewhere near the usage of `EsExecutors.DIRECT_EXECUTOR_SERVICE` here just to confirm that this was a conscious decision?

https://github.com/elastic/elasticsearch/blob/6b85f740cdf00c83e3c2e912455a4aecd7a5e4ff/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java#L215
x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java
ExchangeSinkHandler sinkHandler = new ExchangeSinkHandler(maxBufferSize, threadPool::relativeTimeInMillis);
        if (sinks.putIfAbsent(exchangeId, sinkHandler) != null) {
            throw new IllegalStateException("sink exchanger for id [" + exchangeId + "] already exists");
        }
        return sinkHandler;
    }
    /**
     * Returns an exchange sink handler for the given id.
     */
    public ExchangeSinkHandler getSinkHandler(String exchangeId) {
        ExchangeSinkHandler sinkHandler = sinks.get(exchangeId);
        if (sinkHandler == null) {
            throw new ResourceNotFoundException("sink exchanger for id [{}] doesn't exist", exchangeId);
        }
        return sinkHandler;
    }
    /**
     * Removes the exchange sink handler associated with the given exchange id.
     */
    public void finishSinkHandler(String exchangeId, Exception failure) {
        final ExchangeSinkHandler sinkHandler = sinks.remove(exchangeId);
        if (sinkHandler != null) {
            if (failure != null) {
                sinkHandler.onFailure(failure);
            }
            assert sinkHandler.isFinished() : "Exchange sink " + exchangeId + " wasn't finished yet";
        }
    }
    /**
     * Creates an {@link ExchangeSourceHandler} for the specified exchange id.