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/test/java/org/elasticsearch/compute/operator/exchange/ExchangeServiceTests.java
}
        ExchangeSinkHandler sinkExchanger = new ExchangeSinkHandler(2, threadPool::relativeTimeInMillis);
        ExchangeSink sink1 = sinkExchanger.createExchangeSink();
        ExchangeSink sink2 = sinkExchanger.createExchangeSink();
        ExchangeSourceHandler sourceExchanger = new ExchangeSourceHandler(3, threadPool.executor(ESQL_TEST_EXECUTOR));
        assertThat(sourceExchanger.refCount(), equalTo(1));
        ExchangeSource source = sourceExchanger.createExchangeSource();
        assertThat(sourceExchanger.refCount(), equalTo(2));
        sourceExchanger.addRemoteSink(sinkExchanger::fetchPageAsync, 1);
        assertThat(sourceExchanger.refCount(), equalTo(3));
        ListenableActionFuture<Void> waitForReading = source.waitForReading();
        assertFalse(waitForReading.isDone());
        assertNull(source.pollPage());
        assertTrue(sink1.waitForWriting().isDone());
        randomFrom(sink1, sink2).addPage(pages[0]);
        randomFrom(sink1, sink2).addPage(pages[1]);
        
        for (Page p : List.of(pages[2], pages[3], pages[4])) {
            ExchangeSink sink = randomFrom(sink1, sink2);
            assertBusy(() -> assertTrue(sink.waitForWriting().isDone()));
            sink.addPage(p);
        }
        
        assertFalse(randomFrom(sink1, sink2).waitForWriting().isDone());
        assertBusy(() -> assertTrue(source.waitForReading().isDone()));
        assertEquals(pages[0], source.pollPage());
        assertBusy(() -> assertTrue(source.waitForReading().isDone()));
        assertEquals(pages[1], source.pollPage());
        
        assertBusy(() -> assertTrue(randomFrom(sink1, sink2).waitForWriting().isDone()));