`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
import org.elasticsearch.tasks.TaskCancelledException;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportChannel;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.transport.TransportRequestHandler;
import org.elasticsearch.transport.TransportRequestOptions;
import org.elasticsearch.transport.TransportResponse;
import org.elasticsearch.transport.TransportService;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.Executor;
/**
 * {@link ExchangeService} is responsible for exchanging pages between exchange sinks and sources on the same or different nodes.
 * It holds a map of {@link ExchangeSourceHandler} and {@link ExchangeSinkHandler} instances for each node in the cluster.
 * To connect exchange sources to exchange sinks, use the {@link ExchangeSourceHandler#addRemoteSink(RemoteSink, int)} method.
 */
public final class ExchangeService extends AbstractLifecycleComponent {
    
    
    public static final String EXCHANGE_ACTION_NAME = "internal:data/read/esql/exchange";
    private static final String OPEN_EXCHANGE_ACTION_NAME = "internal:data/read/esql/open_exchange";
    /**
     * The time interval for an exchange sink handler to be considered inactive and subsequently
     * removed from the exchange service if no sinks are attached (i.e., no computation uses that sink handler).
     */
    public static final String INACTIVE_SINKS_INTERVAL_SETTING = "esql.exchange.sink_inactive_interval";
    private static final Logger LOGGER = LogManager.getLogger(ExchangeService.class);
    private final ThreadPool threadPool;
    private final String requestExecutorName;
    private final Executor responseExecutor;