`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
private final String sessionId;
        private final int exchangeBuffer;
        OpenExchangeRequest(String sessionId, int exchangeBuffer) {
            this.sessionId = sessionId;
            this.exchangeBuffer = exchangeBuffer;
        }
        OpenExchangeRequest(StreamInput in) throws IOException {
            super(in);
            this.sessionId = in.readString();
            this.exchangeBuffer = in.readVInt();
        }
        @Override
        public void writeTo(StreamOutput out) throws IOException {
            super.writeTo(out);
            out.writeString(sessionId);
            out.writeVInt(exchangeBuffer);
        }
    }
    private class OpenExchangeRequestHandler implements TransportRequestHandler<OpenExchangeRequest> {