[CCR] Delay auto follow license check (#33557)

* [CCR] Delay auto follow license check
so that we're sure that there are auto follow patterns configured

Otherwise we log a warning in case someone is running with basic or gold
license and has not used the ccr feature.

Author: martijnvg

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

@@ -70,12 +70,6 @@ public AutoFollowCoordinator(
     }
 
     private void doAutoFollow() {
-        if (ccrLicenseChecker.isCcrAllowed() == false) {
-            // TODO: set non-compliant status on auto-follow coordination that can be viewed via a stats API
-            LOGGER.warn("skipping auto-follower coordination", LicenseUtils.newComplianceException("ccr"));
-            threadPool.schedule(pollInterval, ThreadPool.Names.SAME, this::doAutoFollow);
-            return;
-        }
         if (localNodeMaster == false) {
             return;
         }
@@ -91,6 +85,13 @@ private void doAutoFollow() {
             return;
         }
 
+        if (ccrLicenseChecker.isCcrAllowed() == false) {
+            // TODO: set non-compliant status on auto-follow coordination that can be viewed via a stats API
+            LOGGER.warn("skipping auto-follower coordination", LicenseUtils.newComplianceException("ccr"));
+            threadPool.schedule(pollInterval, ThreadPool.Names.SAME, this::doAutoFollow);
+            return;
+        }
+
         Consumer<Exception> handler = e -> {
             if (e != null) {
                 LOGGER.warn("failure occurred during auto-follower coordination", e);

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

@@ -12,6 +12,10 @@
 import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateUpdateTask;
+import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.plugins.Plugin;
@@ -23,6 +27,8 @@
 import org.elasticsearch.xpack.ccr.action.FollowIndexAction;
 import org.elasticsearch.xpack.ccr.action.PutAutoFollowPatternAction;
 import org.elasticsearch.xpack.ccr.action.ShardFollowNodeTask;
+import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata;
+import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata.AutoFollowPattern;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -127,6 +133,40 @@ public void onFailure(final Exception e) {
     }
 
     public void testAutoFollowCoordinatorLogsSkippingAutoFollowCoordinationWithNonCompliantLicense() throws Exception {
+        // Update the cluster state so that we have auto follow patterns and verify that we log a warning in case of incompatible license:
+        CountDownLatch latch = new CountDownLatch(1);
+        ClusterService clusterService = getInstanceFromNode(ClusterService.class);
+        clusterService.submitStateUpdateTask("test-add-auto-follow-pattern", new ClusterStateUpdateTask() {
+
+            @Override
+            public ClusterState execute(ClusterState currentState) throws Exception {
+                AutoFollowPattern autoFollowPattern =
+                    new AutoFollowPattern(Collections.singletonList("logs-*"), null, null, null, null, null, null, null, null);
+                AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata(
+                    Collections.singletonMap("test_alias", autoFollowPattern),
+                    Collections.emptyMap()
+                );
+
+                ClusterState.Builder newState = ClusterState.builder(currentState);
+                newState.metaData(MetaData.builder(currentState.getMetaData())
+                    .putCustom(AutoFollowMetadata.TYPE, autoFollowMetadata)
+                    .build());
+                return newState.build();
+            }
+
+            @Override
+            public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+                latch.countDown();
+            }
+
+            @Override
+            public void onFailure(String source, Exception e) {
+                latch.countDown();
+                fail("unexpected error [" + e.getMessage() + "]");
+            }
+        });
+        latch.await();
+
         final Logger logger = LogManager.getLogger(AutoFollowCoordinator.class);
         final MockLogAppender appender = new MockLogAppender();
         appender.start();