### 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 
server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java
* Used for ccs_minimize_roundtrips=false
     */
    static void collectSearchShards(
        IndicesOptions indicesOptions,
        String preference,
        String routing,
        QueryBuilder query,
        boolean allowPartialResults,
        SearchContextId searchContext,
        Map<String, OriginalIndices> remoteIndicesByCluster,
        SearchResponse.Clusters clusters,
        SearchTimeProvider timeProvider,
        TransportService transportService,
        ActionListener<Map<String, SearchShardsResponse>> listener
    ) {
        RemoteClusterService remoteClusterService = transportService.getRemoteClusterService();
        final CountDown responsesCountDown = new CountDown(remoteIndicesByCluster.size());
        final Map<String, SearchShardsResponse> searchShardsResponses = new ConcurrentHashMap<>();
        final AtomicReference<Exception> exceptions = new AtomicReference<>();
        for (Map.Entry<String, OriginalIndices> entry : remoteIndicesByCluster.entrySet()) {
            final String clusterAlias = entry.getKey();
            boolean skipUnavailable = remoteClusterService.isSkipUnavailable(clusterAlias);
            TransportSearchAction.CCSActionListener<SearchShardsResponse, Map<String, SearchShardsResponse>> singleListener =
                new TransportSearchAction.CCSActionListener<>(
                    clusterAlias,
                    skipUnavailable,
                    responsesCountDown,
                    exceptions,
                    clusters.getCluster(clusterAlias),
                    listener