### 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/main/java/org/elasticsearch/xpack/downsample/TransportDownsampleIndexerAction.java
import org.elasticsearch.action.support.broadcast.TransportBroadcastAction;
import org.elasticsearch.client.internal.Client;
import org.elasticsearch.client.internal.OriginSettingClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.routing.GroupShardsIterator;
import org.elasticsearch.cluster.routing.ShardIterator;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.indices.IndicesService;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.xpack.core.ClientHelper;
import org.elasticsearch.xpack.core.downsample.DownsampleIndexerAction;
import org.elasticsearch.xpack.core.rollup.action.RollupShardTask;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicReferenceArray;
import static org.elasticsearch.xpack.rollup.Rollup.TASK_THREAD_POOL_NAME;
/**
 * A {@link TransportBroadcastAction} that rollups all the shards of a source index into a new rollup index.
 *
 * TODO: Enforce that we don't retry on another replica if we throw an error after sending some buckets.
 */
public class TransportDownsampleIndexerAction extends TransportBroadcastAction<