`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
InactiveSinksReaper(Logger logger, ThreadPool threadPool, TimeValue interval) {
            super(logger, threadPool, EsExecutors.DIRECT_EXECUTOR_SERVICE, interval, true);
            rescheduleIfNecessary();
        }
        @Override
        protected boolean mustReschedule() {
            Lifecycle.State state = lifecycleState();
            return state != Lifecycle.State.STOPPED && state != Lifecycle.State.CLOSED;
        }
        @Override
        protected void runInternal() {
            final TimeValue maxInterval = getInterval();
            final long nowInMillis = threadPool.relativeTimeInMillis();
            for (Map.Entry<String, ExchangeSinkHandler> e : sinks.entrySet()) {
                ExchangeSinkHandler sink = e.getValue();
                if (sink.hasData() && sink.hasListeners()) {
                    continue;
                }
                long elapsed = nowInMillis - sink.lastUpdatedTimeInMillis();
                if (elapsed > maxInterval.millis()) {
                    finishSinkHandler(
                        e.getKey(),
                        new ElasticsearchTimeoutException(