`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
node,
                EXCHANGE_ACTION_NAME,
                new ExchangeRequest(exchangeId, allSourcesFinished),
                parentTask,
                TransportRequestOptions.EMPTY,
                new ActionListenerResponseHandler<>(listener, ExchangeResponse::new, responseExecutor)
            );
        }
    }
    
    public boolean isEmpty() {
        return sources.isEmpty() && sinks.isEmpty();
    }
    @Override
    protected void doStart() {
    }
    @Override
    protected void doStop() {
        inactiveSinksReaper.close();
    }
    @Override
    protected void doClose() {
        doStop();
    }
    @Override
    public String toString() {
        return "ExchangeService{" + "sinks=" + sinks.keySet() + ", sources=" + sources.keySet() + '}';
    }
}