Add messages for CCR on license state changes (#52470)

When a license expires, or license state changes, functionality might be
disabled. This commit adds messages for CCR to inform users that CCR
functionality will be disabled when a license expires, or when license
state changes to a license level lower than trial/platinum/enterprise.

x-pack/plugin/core/src/main/java/org/elasticsearch/license/XPackLicenseState.java

@@ -82,6 +82,13 @@ public class XPackLicenseState {
         messages.put(XPackField.ANALYTICS, new String[] {
             "Aggregations provided by Analytics plugin are no longer usable."
         });
+        messages.put(XPackField.CCR, new String[]{
+            "Creating new follower indices will be blocked",
+            "Configuring auto-follow patterns will be blocked",
+            "Auto-follow patterns will no longer discover new leader indices",
+            "The CCR monitoring endpoint will be blocked",
+            "Existing follower indices will continue to replicate data"
+        });
         EXPIRATION_MESSAGES = Collections.unmodifiableMap(messages);
     }
 
@@ -100,6 +107,7 @@ public class XPackLicenseState {
         messages.put(XPackField.LOGSTASH, XPackLicenseState::logstashAcknowledgementMessages);
         messages.put(XPackField.BEATS, XPackLicenseState::beatsAcknowledgementMessages);
         messages.put(XPackField.SQL, XPackLicenseState::sqlAcknowledgementMessages);
+        messages.put(XPackField.CCR, XPackLicenseState::ccrAcknowledgementMessages);
         ACKNOWLEDGMENT_MESSAGES = Collections.unmodifiableMap(messages);
     }
 
@@ -266,6 +274,26 @@ private static String[] sqlAcknowledgementMessages(OperationMode currentMode, Op
         return Strings.EMPTY_ARRAY;
     }
 
+    private static String[] ccrAcknowledgementMessages(final OperationMode current, final OperationMode next) {
+        switch (current) {
+            // the current license level permits CCR
+            case TRIAL:
+            case PLATINUM:
+                switch (next) {
+                    // the next license level does not permit CCR
+                    case MISSING:
+                    case BASIC:
+                    case STANDARD:
+                    case GOLD:
+                        // so CCR will be disabled
+                        return new String[]{
+                            "Cross-Cluster Replication will be disabled"
+                        };
+                }
+        }
+        return Strings.EMPTY_ARRAY;
+    }
+
     private static boolean isBasic(OperationMode mode) {
         return mode == OperationMode.BASIC;
     }

x-pack/plugin/core/src/test/java/org/elasticsearch/license/XPackLicenseStateTests.java

@@ -491,6 +491,75 @@ public void testSqlAckTrialOrPlatinumToNotTrialOrPlatinum() {
         assertAckMesssages(XPackField.SQL, randomTrialOrPlatinumMode(), randomBasicStandardOrGold(), 1);
     }
 
+    public void testCcrDefaults() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        assertTrue(state.isCcrAllowed());
+    }
+
+    public void testCcrBasic() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(BASIC, true, null);
+
+        assertThat(state.isCcrAllowed(), is(false));
+    }
+
+    public void testCcrBasicExpired() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(BASIC, false, null);
+
+        assertThat(state.isCcrAllowed(), is(false));
+    }
+
+    public void testCcrStandard() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(STANDARD, true, null);
+
+        assertThat(state.isCcrAllowed(), is(false));
+    }
+
+    public void testCcrStandardExpired() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(STANDARD, false, null);
+
+        assertThat(state.isCcrAllowed(), is(false));
+    }
+
+    public void testCcrGold() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(GOLD, true, null);
+
+        assertThat(state.isCcrAllowed(), is(false));
+    }
+
+    public void testCcrGoldExpired() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(GOLD, false, null);
+
+        assertThat(state.isCcrAllowed(), is(false));
+    }
+
+    public void testCcrPlatinum() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(PLATINUM, true, null);
+
+        assertTrue(state.isCcrAllowed());
+    }
+
+    public void testCcrPlatinumExpired() {
+        final XPackLicenseState state = new XPackLicenseState(Settings.EMPTY);
+        state.update(PLATINUM, false, null);
+
+        assertFalse(state.isCcrAllowed());
+    }
+
+    public void testCcrAckAnyToTrialOrPlatinum() {
+        assertAckMesssages(XPackField.CCR, randomMode(), randomTrialOrPlatinumMode(), 0);
+    }
+
+    public void testCcrAckTrialOrPlatinumToNotTrialOrPlatinum() {
+        assertAckMesssages(XPackField.CCR, randomTrialOrPlatinumMode(), randomBasicStandardOrGold(), 1);
+    }
+
     public void testTransformBasic() throws Exception {
         assertAllowed(BASIC, true, XPackLicenseState::isTransformAllowed, true);
     }