In https://github.com/elastic/elasticsearch/pull/78547 we introduced batching for the ILM master tasks that occur on the happy path. However if a high-shard-count cluster encounters problems while doing ILM-related things—perhaps some nodes are temporarily unavailable for taking a snapshot—then we process the resulting `ilm-retry-failed-step` and `ilm-move-to-error-step` tasks one-by-one which can significantly delay the cluster's recovery from its problems.

We should batch these things together too.

It looks like we also enqueue duplicate `ilm-retry-failed-step` on each poll interval too, although we do appear to treat the duplicates as no-ops at execution time.

Relates https://github.com/elastic/elasticsearch/issues/77466
x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunner.java
* Tracks already executing {@link IndexLifecycleClusterStateUpdateTask} tasks in {@link #executingTasks} to prevent queueing up
     * duplicate cluster state updates.
     * TODO: refactor ILM logic so that this is not required any longer. It is unreasonably expensive to only filter out duplicate tasks at
     *       this point given how these tasks are mostly set up on the cluster state applier thread.
     *
     * @param source source string as used in {@link ClusterService#submitStateUpdateTask(String, ClusterStateTaskConfig,
     *               ClusterStateTaskExecutor)}
     * @param task   task to submit unless already tracked in {@link #executingTasks}.
     */
    private void submitUnlessAlreadyQueued(String source, IndexLifecycleClusterStateUpdateTask task) {
        if (executingTasks.add(task)) {
            final Tuple<Index, StepKey> dedupKey = Tuple.tuple(task.index, task.currentStepKey);
            
            busyIndices.add(dedupKey);
            task.addListener(ActionListener.wrap(() -> {
                final boolean removed = executingTasks.remove(task);
                busyIndices.remove(dedupKey);
                assert removed : "tried to unregister unknown task [" + task + "]";
            }));
            clusterService.submitStateUpdateTask(source, task, ILM_TASK_CONFIG, ILM_TASK_EXECUTOR);
        } else {
            logger.trace("skipped redundant execution of [{}]", source);
        }
    }
    @SuppressForbidden(reason = "legacy usage of unbatched task") 
    private static <T extends ClusterStateUpdateTask> ClusterStateTaskExecutor<T> newExecutor() {
        return ClusterStateTaskExecutor.unbatched();
    }
}