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
@Override
        @SuppressForbidden(reason = "consuming published cluster state for legacy reasons")
        public ClusterState execute(BatchExecutionContext<CloseIndicesTask> batchExecutionContext) {
            var listener = new AllocationActionMultiListener<CloseIndexResponse>(threadPool.getThreadContext());
            var state = batchExecutionContext.initialState();
            for (final var taskContext : batchExecutionContext.taskContexts()) {
                final var task = taskContext.getTask();
                try {
                    final Tuple<ClusterState, List<IndexResult>> closingResult = closeRoutingTable(
                        state,
                        task.blockedIndices,
                        task.verifyResults
                    );
                    state = closingResult.v1();
                    final List<IndexResult> indices = closingResult.v2();
                    assert indices.size() == task.verifyResults.size();
                    taskContext.success(clusterState -> {
                        final boolean acknowledged = indices.stream().noneMatch(IndexResult::hasFailures);
                        final String[] waitForIndices = indices.stream()
                            .filter(result -> result.hasFailures() == false)
                            .filter(result -> clusterState.routingTable().hasIndex(result.getIndex()))
                            .map(result -> result.getIndex().getName())
                            .toArray(String[]::new);
                        if (waitForIndices.length > 0) {
                            ActiveShardsObserver.waitForActiveShards(