### 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
.setSettings(Settings.builder().put(IndexMetadata.INDEX_BLOCKS_WRITE_SETTING.getKey(), true).build())
                .get()
        );
    }
    private void rollup(String sourceIndex, String rollupIndex, DownsampleConfig config) {
        assertAcked(
            client().execute(DownsampleAction.INSTANCE, new DownsampleAction.Request(sourceIndex, rollupIndex, config)).actionGet()
        );
    }
    private RolloverResponse rollover(String dataStreamName) throws ExecutionException, InterruptedException {
        RolloverResponse response = client().admin().indices().rolloverIndex(new RolloverRequest(dataStreamName, null)).get();
        assertAcked(response);
        return response;
    }
    private Aggregations aggregate(final String index, AggregationBuilder aggregationBuilder) {
        return client().prepareSearch(index).addAggregation(aggregationBuilder).get().getAggregations();
    }
    @SuppressWarnings("unchecked")
    private void assertRollupIndex(String sourceIndex, String rollupIndex, DownsampleConfig config) throws IOException {
        
        final GetMappingsResponse getMappingsResponse = client().admin().indices().prepareGetMappings(sourceIndex).get();
        final Map<String, Object> sourceIndexMappings = getMappingsResponse.mappings()
            .entrySet()
            .stream()
            .filter(entry -> sourceIndex.equals(entry.getKey()))
            .findFirst()
            .map(mappingMetadata -> mappingMetadata.getValue().sourceAsMap())
            .orElseThrow(() -> new IllegalArgumentException("No mapping found for rollup source index [" + sourceIndex + "]"));
        final IndexMetadata indexMetadata = client().admin().cluster().prepareState().get().getState().getMetadata().index(sourceIndex);
        final IndicesService indicesService = getInstanceFromNode(IndicesService.class);