### 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
prepareSourceIndex(sourceIndex);
        rollup(sourceIndex, rollupIndex, config);
        assertRollupIndex(sourceIndex, rollupIndex, config);
    }
    public void testCannotRollupWriteableIndex() {
        DownsampleConfig config = new DownsampleConfig(randomInterval());
        
        Exception exception = expectThrows(ElasticsearchException.class, () -> rollup(sourceIndex, rollupIndex, config));
        assertThat(exception.getMessage(), containsString("Rollup requires setting [index.blocks.write = true] for index"));
    }
    public void testCannotRollupMissingIndex() {
        DownsampleConfig config = new DownsampleConfig(randomInterval());
        IndexNotFoundException exception = expectThrows(IndexNotFoundException.class, () -> rollup("missing-index", rollupIndex, config));
        assertEquals("missing-index", exception.getIndex().getName());
        assertThat(exception.getMessage(), containsString("no such index [missing-index]"));
    }
    public void testCannotRollupWhileOtherRollupInProgress() throws Exception {
        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);
        var rollupListener = new ActionListener<AcknowledgedResponse>() {
            boolean success;
            @Override
            public void onResponse(AcknowledgedResponse acknowledgedResponse) {