### 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
protected Collection<Class<? extends Plugin>> getPlugins() {
        return List.of(
            LocalStateCompositeXPackPlugin.class,
            Rollup.class,
            AggregateMetricMapperPlugin.class,
            DataStreamsPlugin.class,
            IndexLifecycle.class
        );
    }
    @Before
    public void setup() throws IOException {
        sourceIndex = randomAlphaOfLength(14).toLowerCase(Locale.ROOT);
        rollupIndex = "rollup-" + sourceIndex;
        startTime = randomLongBetween(946769284000L, 1607470084000L); 
        docCount = randomIntBetween(10, 9000);
        numOfShards = randomIntBetween(1, 4);
        numOfReplicas = randomIntBetween(0, 3);
        
        dimensionValues = new ArrayList<>(MAX_DIM_VALUES);
        for (int j = 0; j < randomIntBetween(1, MAX_DIM_VALUES); j++) {
            dimensionValues.add(randomAlphaOfLength(6));
        }
        /**
         * NOTE: here we map each numeric label field also as a (counter) metric.
         * This is done for testing purposes. There is no easy way to test
         * that labels are collected using the last value. The idea is to
         * check that the value of the label (last value) matches the value
         * of the corresponding metric which uses a last_value metric type.
         */
        Settings.Builder settings = indexSettings(numOfShards, numOfReplicas).put(IndexSettings.MODE.getKey(), IndexMode.TIME_SERIES)