`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
*
     * @throws IllegalStateException if a source handler for the given id already exists
     */
    public ExchangeSourceHandler createSourceHandler(String exchangeId, int maxBufferSize, String fetchExecutor) {
        ExchangeSourceHandler sourceHandler = new ExchangeSourceHandler(maxBufferSize, threadPool.executor(fetchExecutor));
        if (sources.putIfAbsent(exchangeId, sourceHandler) != null) {
            throw new IllegalStateException("source exchanger for id [" + exchangeId + "] already exists");
        }
        sourceHandler.addCompletionListener(ActionListener.releasing(() -> sources.remove(exchangeId)));
        return sourceHandler;
    }
    /**
     * Opens a remote sink handler on the remote node for the given session ID.
     */
    public static void openExchange(
        TransportService transportService,
        DiscoveryNode targetNode,
        String sessionId,
        int exchangeBuffer,
        Executor responseExecutor,
        ActionListener<Void> listener
    ) {
        transportService.sendRequest(
            targetNode,
            OPEN_EXCHANGE_ACTION_NAME,
            new OpenExchangeRequest(sessionId, exchangeBuffer),
            new ActionListenerResponseHandler<>(listener.map(unused -> null), in -> TransportResponse.Empty.INSTANCE, responseExecutor)
        );
    }
    private static class OpenExchangeRequest extends TransportRequest {