### 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
if (acknowledgedResponse.isAcknowledged()) {
                    success = true;
                } else {
                    fail("Failed to receive rollup acknowledgement");
                }
            }
            @Override
            public void onFailure(Exception e) {
                fail("Rollup failed: " + e.getMessage());
            }
        };
        client().execute(DownsampleAction.INSTANCE, new DownsampleAction.Request(sourceIndex, rollupIndex, config), rollupListener);
        assertBusy(() -> {
            try {
                assertEquals(client().admin().indices().prepareGetIndex().addIndices(rollupIndex).get().getIndices().length, 1);
            } catch (IndexNotFoundException e) {
                fail("rollup index has not been created");
            }
        });
        ResourceAlreadyExistsException exception = expectThrows(
            ResourceAlreadyExistsException.class,
            () -> rollup(sourceIndex, rollupIndex, config)
        );
        assertThat(exception.getMessage(), containsString(rollupIndex));
        
        assertBusy(() -> assertTrue("In progress rollup did not complete", rollupListener.success), 60, TimeUnit.SECONDS);
    }
    public void testRollupDatastream() throws Exception {
        DownsampleConfig config = new DownsampleConfig(randomInterval());
        String dataStreamName = createDataStream();