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
import org.elasticsearch.tasks.Task;
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.TransportResponseHandler;
import org.elasticsearch.transport.TransportService;
import java.io.IOException;
import java.util.Map;
/**
 * {@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 Map<String, ExchangeSinkHandler> sinks = ConcurrentCollections.newConcurrentMap();