Performing any long running task or CPU intensive work on transport_worker threads is problematic. This is typically very bad for cluster stability.

This issue has been filed to examine the usage of transport_worked threads by ESQL, with the intention of "offloading" as much as possible, including deserialising large(ish) amounts of data.
x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java
private final Map<String, ExchangeSourceHandler> sources = ConcurrentCollections.newConcurrentMap();
    private final InactiveSinksReaper inactiveSinksReaper;
    public ExchangeService(Settings settings, ThreadPool threadPool) {
        this.threadPool = threadPool;
        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.Names.SAME,
            ExchangeRequest::new,
            new ExchangeTransportAction()
        );
        transportService.registerRequestHandler(
            OPEN_EXCHANGE_ACTION_NAME,
            ThreadPool.Names.SAME,
            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) {
        ExchangeSinkHandler sinkHandler = new ExchangeSinkHandler(maxBufferSize, threadPool::relativeTimeInMillis);
        if (sinks.putIfAbsent(exchangeId, sinkHandler) != null) {
            throw new IllegalStateException("sink exchanger for id [" + exchangeId + "] already exists");