### Description

`SearchResponse.Clusters` currently uses a `HashMap<String, AtomicReference>` for storing Cluster objects. The AtomicReference CAS swaps are done in a do/while loop in most cases (e.g., in `CCSSingleCoordinatorSearchProgressListener`).

This can be replaced with a `ConcurrentHashMap` in conjunction with either `compute()` method or the `replace(key, assumed_previous_value, value)` method for atomically swapping in a new Cluster object with a CAS operation, assuring that the updates don't lose data from competing threads doing separate Cluster object updates.

See: https://github.com/elastic/elasticsearch/pull/98457/files#r1310467586 
x-pack/plugin/async-search/src/internalClusterTest/java/org/elasticsearch/xpack/search/CrossClusterAsyncSearchIT.java
assertFalse("taskInfo on remote cluster should not be cancelled yet: " + taskInfo, taskInfo.cancelled());
            }
            final CancelTasksRequest cancelRequest = new CancelTasksRequest().setTargetTaskId(rootTask.taskId());
            cancelRequest.setWaitForCompletion(randomBoolean());
            cancelFuture = client().admin().cluster().cancelTasks(cancelRequest);
            assertBusy(() -> {
                final Iterable<TransportService> transportServices = cluster(REMOTE_CLUSTER).getInstances(TransportService.class);
                for (TransportService transportService : transportServices) {
                    Collection<CancellableTask> cancellableTasks = transportService.getTaskManager().getCancellableTasks().values();
                    for (CancellableTask cancellableTask : cancellableTasks) {
                        if (cancellableTask.getAction().contains(SearchAction.INSTANCE.name())) {
                            assertTrue(cancellableTask.getDescription(), cancellableTask.isCancelled());
                        }
                    }
                }
            });
            List<TaskInfo> remoteSearchTasksAfterCancellation = client(REMOTE_CLUSTER).admin()
                .cluster()
                .prepareListTasks()
                .get()
                .getTasks()
                .stream()
                .filter(t -> t.action().contains(SearchAction.INSTANCE.name()))
                .toList();
            for (TaskInfo taskInfo : remoteSearchTasksAfterCancellation) {
                assertTrue(taskInfo.description(), taskInfo.cancelled());
            }
            
            AsyncSearchResponse searchResponseAfterCancellation = getAsyncSearch(response.getId());
            assertTrue(searchResponseAfterCancellation.isPartial());