`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 Map<String, ExchangeSinkHandler> sinks = ConcurrentCollections.newConcurrentMap();
    private final Map<String, ExchangeSourceHandler> sources = ConcurrentCollections.newConcurrentMap();
    private final InactiveSinksReaper inactiveSinksReaper;
    public ExchangeService(Settings settings, ThreadPool threadPool, String executorName) {
        this.threadPool = threadPool;
        this.requestExecutorName = executorName;
        this.responseExecutor = threadPool.executor(executorName);
        final var inactiveInterval = settings.getAsTime(INACTIVE_SINKS_INTERVAL_SETTING, TimeValue.timeValueMinutes(5));
        this.inactiveSinksReaper = new InactiveSinksReaper(LOGGER, threadPool, inactiveInterval);
    }
    public void registerTransportHandler(TransportService transportService) {
        transportService.registerRequestHandler(
            EXCHANGE_ACTION_NAME,
            threadPool.executor(requestExecutorName),
            ExchangeRequest::new,
            new ExchangeTransportAction()
        );
        transportService.registerRequestHandler(
            OPEN_EXCHANGE_ACTION_NAME,
            threadPool.executor(requestExecutorName),
            OpenExchangeRequest::new,
            new OpenExchangeRequestHandler()
        );
    }
    /**
     * Creates an {@link ExchangeSinkHandler} for the specified exchange id.
     *
     * @throws IllegalStateException if a sink handler for the given id already exists
     */
    ExchangeSinkHandler createSinkHandler(String exchangeId, int maxBufferSize) {