Deduplicate Heavy CCR Repository CS Requests (#91398) (#95372)

We run the same request back to back for each put-follower call during
the restore. Also, concurrent put-follower calls will all run the same
full CS request concurrently.
In older versions prior to #87235
the concurrency was limited by the size of the snapshot pool. With that
fix though, they are run at almost arbitry concurrency when many
put-follow requests are executed concurrently.
-> fixed by using the existing deduplicator to only run a single remote
CS request at a time for each CCR repository.
Also, this removes the needless forking in the put-follower action that
is not necessary any longer now that we have the CCR repository
non-blocking (we do the same for normal restores that can safely be
started from a transport thread), which should fix some bad-ux
situations where the snapshot threads are busy on master, making
the put-follower requests not go through in time.

Author: original-brownbear

docs/changelog/91398.yaml

@@ -0,0 +1,5 @@
+pr: 91398
+summary: Deduplicate Heavy CCR Repository CS Requests
+area: CCR
+type: bug
+issues: []

server/src/test/java/org/elasticsearch/transport/SingleResultDeduplicatorTests.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.transport;
+
+import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.SingleResultDeduplicator;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.test.ESTestCase;
+
+public class SingleResultDeduplicatorTests extends ESTestCase {
+
+    public void testDeduplicatesWithoutShowingStaleData() {
+        final SetOnce<ActionListener<Object>> firstListenerRef = new SetOnce<>();
+        final SetOnce<ActionListener<Object>> secondListenerRef = new SetOnce<>();
+        final SingleResultDeduplicator<Object> deduplicator = new SingleResultDeduplicator<>(new ThreadContext(Settings.EMPTY), l -> {
+            if (firstListenerRef.trySet(l) == false) {
+                secondListenerRef.set(l);
+            }
+        });
+        final Object result1 = new Object();
+        final Object result2 = new Object();
+
+        final int totalListeners = randomIntBetween(2, 10);
+        final boolean[] called = new boolean[totalListeners];
+        deduplicator.execute(new ActionListener<Object>() {
+            @Override
+            public void onResponse(Object response) {
+                assertFalse(called[0]);
+                called[0] = true;
+                assertEquals(result1, response);
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                throw new AssertionError(e);
+            }
+        });
+
+        for (int i = 1; i < totalListeners; i++) {
+            final int index = i;
+            deduplicator.execute(new ActionListener<Object>() {
+
+                @Override
+                public void onResponse(Object response) {
+                    assertFalse(called[index]);
+                    called[index] = true;
+                    assertEquals(result2, response);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    throw new AssertionError(e);
+                }
+            });
+        }
+        for (int i = 0; i < totalListeners; i++) {
+            assertFalse(called[i]);
+        }
+        firstListenerRef.get().onResponse(result1);
+        assertTrue(called[0]);
+        for (int i = 1; i < totalListeners; i++) {
+            assertFalse(called[i]);
+        }
+        secondListenerRef.get().onResponse(result2);
+        for (int i = 0; i < totalListeners; i++) {
+            assertTrue(called[i]);
+        }
+    }
+}

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionRunnable;
+import org.elasticsearch.action.SingleResultDeduplicator;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
@@ -143,6 +144,8 @@ public class CcrRepository extends AbstractLifecycleComponent implements Reposit
 
     private final CounterMetric throttledTime = new CounterMetric();
 
+    private final SingleResultDeduplicator<ClusterState> csDeduplicator;
+
     public CcrRepository(
         RepositoryMetadata metadata,
         Client client,
@@ -159,6 +162,17 @@ public CcrRepository(
         this.ccrLicenseChecker = ccrLicenseChecker;
         this.client = client;
         this.threadPool = threadPool;
+        csDeduplicator = new SingleResultDeduplicator<>(
+            threadPool.getThreadContext(),
+            l -> getRemoteClusterClient().admin()
+                .cluster()
+                .prepareState()
+                .clear()
+                .setMetadata(true)
+                .setNodes(true)
+                .setMasterNodeTimeout(TimeValue.MAX_VALUE)
+                .execute(l.map(ClusterStateResponse::getState))
+        );
     }
 
     @Override
@@ -191,27 +205,22 @@ public void getSnapshotInfo(GetSnapshotInfoContext context) {
         assert snapshotIds.size() == 1 && SNAPSHOT_ID.equals(snapshotIds.iterator().next())
             : "RemoteClusterRepository only supports " + SNAPSHOT_ID + " as the SnapshotId but saw " + snapshotIds;
         try {
-            getRemoteClusterClient().admin()
-                .cluster()
-                .prepareState()
-                .clear()
-                .setMetadata(true)
-                .setNodes(true)
-                // fork to the snapshot meta pool because the context expects to run on it and asserts that it does
-                .execute(new ThreadedActionListener<>(logger, threadPool, ThreadPool.Names.SNAPSHOT_META, context.map(response -> {
-                    Metadata responseMetadata = response.getState().metadata();
+            csDeduplicator.execute(
+                new ThreadedActionListener<>(logger, threadPool, ThreadPool.Names.SNAPSHOT_META, context.map(response -> {
+                    Metadata responseMetadata = response.metadata();
                     ImmutableOpenMap<String, IndexMetadata> indicesMap = responseMetadata.indices();
-                    List<String> indices = new ArrayList<>(indicesMap.keySet());
                     return new SnapshotInfo(
                         new Snapshot(this.metadata.name(), SNAPSHOT_ID),
-                        indices,
-                        new ArrayList<>(responseMetadata.dataStreams().keySet()),
-                        Collections.emptyList(),
-                        response.getState().getNodes().getMaxNodeVersion(),
+                        org.elasticsearch.core.List.copyOf(indicesMap.keySet()),
+                        org.elasticsearch.core.List.copyOf(responseMetadata.dataStreams().keySet()),
+                        org.elasticsearch.core.List.of(),
+                        response.getNodes().getMaxNodeVersion(),
                         SnapshotState.SUCCESS
                     );
-                }), false));
+                }), false)
+            );
         } catch (Exception e) {
+            assert false : e;
             context.onFailure(e);
         }
     }
@@ -275,8 +284,8 @@ public IndexMetadata getSnapshotIndexMetaData(RepositoryData repositoryData, Sna
     @Override
     public void getRepositoryData(ActionListener<RepositoryData> listener) {
         try {
-            getRemoteClusterClient().admin().cluster().prepareState().clear().setMetadata(true).execute(listener.map(response -> {
-                final Metadata remoteMetadata = response.getState().getMetadata();
+            csDeduplicator.execute(listener.map(response -> {
+                final Metadata remoteMetadata = response.getMetadata();
                 final String[] concreteAllIndices = remoteMetadata.getConcreteAllIndices();
                 final Map<String, SnapshotId> copiedSnapshotIds = new HashMap<>();
                 final Map<String, RepositoryData.SnapshotDetails> snapshotsDetails = new HashMap<>();
@@ -308,6 +317,7 @@ public void getRepositoryData(ActionListener<RepositoryData> listener) {
                 );
             }));
         } catch (Exception e) {
+            assert false;
             listener.onFailure(e);
         }
     }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/repository/CcrRepositoryRetentionLeaseTests.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.seqno.RetentionLeaseActions;
 import org.elasticsearch.index.seqno.RetentionLeaseAlreadyExistsException;
@@ -55,14 +56,7 @@ public void testWhenRetentionLeaseAlreadyExistsWeTryToRenewIt() {
             CcrSettings.getSettings().stream().filter(Setting::hasNodeScope)
         ).collect(Collectors.toSet());
 
-        final CcrRepository repository = new CcrRepository(
-            repositoryMetadata,
-            mock(Client.class),
-            new CcrLicenseChecker(() -> true, () -> true),
-            Settings.EMPTY,
-            new CcrSettings(Settings.EMPTY, new ClusterSettings(Settings.EMPTY, settings)),
-            mock(ThreadPool.class)
-        );
+        final CcrRepository repository = createCcrRepository(repositoryMetadata, settings);
 
         final ShardId followerShardId = new ShardId(new Index("follower-index-name", "follower-index-uuid"), 0);
         final ShardId leaderShardId = new ShardId(new Index("leader-index-name", "leader-index-uuid"), 0);
@@ -112,6 +106,20 @@ public void testWhenRetentionLeaseAlreadyExistsWeTryToRenewIt() {
         verifyNoMoreInteractions(remoteClient);
     }
 
+    private static CcrRepository createCcrRepository(RepositoryMetadata repositoryMetadata, Set<Setting<?>> settings) {
+        final ThreadPool threadPool = mock(ThreadPool.class);
+        final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
+        when(threadPool.getThreadContext()).thenReturn(threadContext);
+        return new CcrRepository(
+            repositoryMetadata,
+            mock(Client.class),
+            new CcrLicenseChecker(() -> true, () -> true),
+            Settings.EMPTY,
+            new CcrSettings(Settings.EMPTY, new ClusterSettings(Settings.EMPTY, settings)),
+            threadPool
+        );
+    }
+
     public void testWhenRetentionLeaseExpiresBeforeWeCanRenewIt() {
         final RepositoryMetadata repositoryMetadata = mock(RepositoryMetadata.class);
         when(repositoryMetadata.name()).thenReturn(CcrRepository.NAME_PREFIX);
@@ -120,14 +128,7 @@ public void testWhenRetentionLeaseExpiresBeforeWeCanRenewIt() {
             CcrSettings.getSettings().stream().filter(Setting::hasNodeScope)
         ).collect(Collectors.toSet());
 
-        final CcrRepository repository = new CcrRepository(
-            repositoryMetadata,
-            mock(Client.class),
-            new CcrLicenseChecker(() -> true, () -> true),
-            Settings.EMPTY,
-            new CcrSettings(Settings.EMPTY, new ClusterSettings(Settings.EMPTY, settings)),
-            mock(ThreadPool.class)
-        );
+        final CcrRepository repository = createCcrRepository(repositoryMetadata, settings);
 
         final ShardId followerShardId = new ShardId(new Index("follower-index-name", "follower-index-uuid"), 0);
         final ShardId leaderShardId = new ShardId(new Index("leader-index-name", "leader-index-uuid"), 0);