### Description

This is related to #96757.

Right now downsampling uses a thread pool with a single running thread (disallowing concurrent downsampling operations at the transport level) and an unbounded queue of waiting threads, queueing and waiting to be executed. We need to explore options around using multiple running threads here.
x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java
new String[] { FIELD_NUMERIC_1, FIELD_NUMERIC_2 },
            new String[] {}
        );
        TaskCancelledException exception = expectThrows(TaskCancelledException.class, () -> indexer.execute());
        assertThat(exception.getMessage(), equalTo("Shard [" + sourceIndex + "][" + shardNum + "] rollup cancelled"));
    }
    public void testRollupBulkFailed() throws IOException {
        
        DownsampleConfig config = new DownsampleConfig(randomInterval());
        SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder()
            .startObject()
            .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval()))
            .field(FIELD_DIMENSION_1, randomAlphaOfLength(1))
            .field(FIELD_NUMERIC_1, randomDouble())
            .endObject();
        bulkIndex(sourceSupplier);
        prepareSourceIndex(sourceIndex);
        
        assertAcked(
            client().admin()
                .indices()
                .preparePutTemplate(rollupIndex)
                .setPatterns(List.of(rollupIndex))
                .setSettings(Settings.builder().put("index.blocks.write", "true").build())
                .get()
        );
        ElasticsearchException exception = expectThrows(ElasticsearchException.class, () -> rollup(sourceIndex, rollupIndex, config));
        assertThat(exception.getMessage(), equalTo("Unable to rollup index [" + sourceIndex + "]"));
    }
    public void testTooManyBytesInFlight() throws IOException {