Retry on wait_for_metada_version timeout (#38521)

Closes #37807
Backport of #38521

Author: dnhatn

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrRequests.java

@@ -57,22 +57,29 @@ public static void getIndexMetadata(Client client, Index index, long mappingVers
         }
         client.admin().cluster().state(request, ActionListener.wrap(
             response -> {
-                if (response.getState() == null) {
+                if (response.getState() == null) { // timeout on wait_for_metadata_version
                     assert metadataVersion > 0 : metadataVersion;
-                    throw new IllegalStateException("timeout to get cluster state with" +
-                        " metadata version [" + metadataVersion + "], mapping version [" + mappingVersion + "]");
+                    if (timeoutSupplier.get().nanos() < 0) {
+                        listener.onFailure(new IllegalStateException("timeout to get cluster state with" +
+                            " metadata version [" + metadataVersion + "], mapping version [" + mappingVersion + "]"));
+                    } else {
+                        getIndexMetadata(client, index, mappingVersion, metadataVersion, timeoutSupplier, listener);
+                    }
+                } else {
+                    final MetaData metaData = response.getState().metaData();
+                    final IndexMetaData indexMetaData = metaData.getIndexSafe(index);
+                    if (indexMetaData.getMappingVersion() >= mappingVersion) {
+                        listener.onResponse(indexMetaData);
+                        return;
+                    }
+                    if (timeoutSupplier.get().nanos() < 0) {
+                        listener.onFailure(new IllegalStateException(
+                            "timeout to get cluster state with mapping version [" + mappingVersion + "]"));
+                    } else {
+                        // ask for the next version.
+                        getIndexMetadata(client, index, mappingVersion, metaData.version() + 1, timeoutSupplier, listener);
+                    }
                 }
-                final MetaData metaData = response.getState().metaData();
-                final IndexMetaData indexMetaData = metaData.getIndexSafe(index);
-                if (indexMetaData.getMappingVersion() >= mappingVersion) {
-                    listener.onResponse(indexMetaData);
-                    return;
-                }
-                if (timeoutSupplier.get().nanos() < 0) {
-                    throw new IllegalStateException("timeout to get cluster state with mapping version [" + mappingVersion + "]");
-                }
-                // ask for the next version.
-                getIndexMetadata(client, index, mappingVersion, metaData.version() + 1, timeoutSupplier, listener);
             },
             listener::onFailure
         ));

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

@@ -235,7 +235,6 @@ public void testAddNewReplicasOnFollower() throws Exception {
         pauseFollow("follower-index");
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/37807")
     public void testReadRequestsReturnLatestMappingVersion() throws Exception {
         InternalTestCluster leaderCluster = getLeaderCluster();
         Settings nodeAttributes = Settings.builder().put("node.attr.box", "large").build();