https://github.com/elastic/elasticsearch/blob/099fb84cfcf87be6689c3516674f45438bafa767/server/src/main/java/org/elasticsearch/action/ActionListener.java#L420-L432

Rather than keeping hold of the `delegate` and using an `AtomicBoolean` to track the completion, we could use an `AtomicReference` and drop the whole listener on completion, which might free up heap in some fail-fast cases.
server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexStateService.java
new AddBlocksTask(request, listener),
            ClusterStateTaskConfig.build(Priority.URGENT, request.masterNodeTimeout()),
            addBlocksExecutor
        );
    }
    private class AddBlocksExecutor extends SimpleBatchedExecutor<AddBlocksTask, Map<Index, ClusterBlock>> {
        @Override
        public Tuple<ClusterState, Map<Index, ClusterBlock>> executeTask(AddBlocksTask task, ClusterState clusterState) {
            return addIndexBlock(task.request.indices(), clusterState, task.request.getBlock());
        }
        @Override
        public void taskSucceeded(AddBlocksTask task, Map<Index, ClusterBlock> blockedIndices) {
            if (blockedIndices.isEmpty()) {
                task.listener().onResponse(AddIndexBlockResponse.EMPTY);
            } else {
                threadPool.executor(ThreadPool.Names.MANAGEMENT)
                    .execute(
                        new WaitForBlocksApplied(
                            blockedIndices,
                            task.request,
                            task.listener().delegateFailure((delegate2, verifyResults) -> {
                                clusterService.submitStateUpdateTask(
                                    "finalize-index-block-["
                                        + task.request.getBlock().name
                                        + "]-["
                                        + blockedIndices.keySet().stream().map(Index::getName).collect(Collectors.joining(", "))
                                        + "]",
                                    new FinalizeBlocksTask(task.request, blockedIndices, verifyResults, delegate2),
                                    ClusterStateTaskConfig.build(Priority.URGENT),
                                    finalizeBlocksExecutor