implement a workaround for remote cluster validation (#50460)

In 7.x an internal API used for validating remote cluster does not throw, see #50420 for the 
details. This change implements a workaround for remote cluster validation, only for 7.x branches.

fixes #50420

server/src/main/java/org/elasticsearch/transport/NoSuchRemoteClusterException.java

@@ -30,7 +30,7 @@ import java.io.IOException;
  */
 public final class NoSuchRemoteClusterException extends ResourceNotFoundException {
 
-    NoSuchRemoteClusterException(String clusterName) {
+    public NoSuchRemoteClusterException(String clusterName) {
         //No node available for cluster
         super("no such remote cluster: [" + clusterName + "]");
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.ClusterNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
@@ -22,6 +23,7 @@ import org.elasticsearch.indices.InvalidIndexNameException;
 import org.elasticsearch.license.RemoteClusterLicenseChecker;
 import org.elasticsearch.protocol.xpack.license.LicenseStatus;
 import org.elasticsearch.transport.NoSuchRemoteClusterException;
+import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.RemoteClusterService;
 
 import java.text.MessageFormat;
@@ -32,6 +34,7 @@ import java.util.Locale;
 import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeSet;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.action.ValidateActions.addValidationError;
 import static org.elasticsearch.cluster.metadata.MetaDataCreateIndexService.validateIndexOrAliasName;
@@ -60,6 +63,9 @@ public final class SourceDestValidator {
     public static final String REMOTE_CLUSTER_LICENSE_INACTIVE = "License check failed for remote cluster "
         + "alias [{0}], license is not active";
 
+    // workaround for 7.x: remoteClusterAliases does not throw
+    private static final ClusterNameExpressionResolver clusterNameExpressionResolver = new ClusterNameExpressionResolver();
+
     private final IndexNameExpressionResolver indexNameExpressionResolver;
     private final RemoteClusterService remoteClusterService;
     private final RemoteClusterLicenseChecker remoteClusterLicenseChecker;
@@ -375,7 +381,7 @@ public final class SourceDestValidator {
             // this can throw
             List<String> remoteAliases;
             try {
-                remoteAliases = RemoteClusterLicenseChecker.remoteClusterAliases(context.getRegisteredRemoteClusterNames(), remoteIndices);
+                remoteAliases = remoteClusterAliases(context.getRegisteredRemoteClusterNames(), remoteIndices);
             } catch (NoSuchRemoteClusterException e) {
                 context.addValidationError(e.getMessage());
                 listener.onResponse(context);
@@ -457,4 +463,27 @@ public final class SourceDestValidator {
     private static String getMessage(String message, Object... args) {
         return new MessageFormat(message, Locale.ROOT).format(args);
     }
+
+    /**
+     * Workaround for 7.x: remoteClusterAliases does not throw
+     *
+     * copied from {@link RemoteClusterLicenseChecker#remoteClusterAliases}, creates a NoSuchRemoteClusterException iff
+     * {@link ClusterNameExpressionResolver#resolveClusterNames} returns an empty list (throws in 8.x).
+     */
+    private static List<String> remoteClusterAliases(final Set<String> remoteClusters, final List<String> indices) {
+        return indices.stream()
+                .filter(RemoteClusterLicenseChecker::isRemoteIndex)
+                .map(index -> index.substring(0, index.indexOf(RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR)))
+                .distinct()
+                .flatMap(clusterExpression ->
+                {
+                    List<String> resolved = clusterNameExpressionResolver.resolveClusterNames(remoteClusters, clusterExpression);
+                    if (resolved.isEmpty()) {
+                        throw new NoSuchRemoteClusterException(clusterExpression);
+                    }
+                    return resolved.stream();
+                })
+                .distinct()
+                .collect(Collectors.toList());
+    }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java

@@ -706,7 +706,6 @@ public class SourceDestValidatorTests extends ESTestCase {
         }, null);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/50420")
     public void testRemoteSourceDoesNotExist() throws InterruptedException {
         Context context = spy(
             new SourceDestValidator.Context(