Backport rolling upgrade multi cluster module to 6.7 branch (#38888)

* Add rolling upgrade multi cluster test module (#38277)

This test starts 2 clusters, each with 3 nodes.
First the leader cluster is started and tests are run against it and
then the follower cluster is started and tests execute against this two cluster.

Then the follower cluster is upgraded, one node at a time.
After that the leader cluster is upgraded, one node at a time.
Every time a node is upgraded tests are ran while both clusters are online.
(and either leader cluster has mixed node versions or the follower cluster)

This commit only tests CCR index following, but could be used for CCS tests as well.
In particular for CCR, unidirectional index following is tested during a rolling upgrade.
During the test several indices are created and followed in the leader cluster before or
while the follower cluster is being upgraded.

This tests also verifies that attempting to follow an index in the upgraded cluster
from the not upgraded cluster fails. After both clusters are upgraded following the
index that previously failed should succeed.

Relates to #37231 and #38037

* Filter out upgraded version index settings when starting index following (#38838)

The `index.version.upgraded` and `index.version.upgraded_string` are likely
to be different between leader and follower index. In the event that
a follower index gets restored on a upgraded node while the leader index
is still on non-upgraded nodes.

Closes #38835

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

@@ -242,7 +242,8 @@ static void validate(
         Settings leaderSettings = filter(leaderIndex.getSettings());
         Settings followerSettings = filter(followIndex.getSettings());
         if (leaderSettings.equals(followerSettings) == false) {
-            throw new IllegalArgumentException("the leader and follower index settings must be identical");
+            throw new IllegalArgumentException("the leader index setting[" + leaderSettings + "] and follower index settings [" +
+                followerSettings + "] must be identical");
         }
 
         // Validates if the current follower mapping is mergable with the leader mapping.
@@ -452,6 +453,11 @@ static Settings filter(Settings originalSettings) {
         settings.remove(IndexMetaData.SETTING_INDEX_PROVIDED_NAME);
         settings.remove(IndexMetaData.SETTING_CREATION_DATE);
 
+        // Follower index may be upgraded, while the leader index hasn't been upgraded, so it is expected
+        // that these settings are different:
+        settings.remove(IndexMetaData.SETTING_VERSION_UPGRADED);
+        settings.remove(IndexMetaData.SETTING_VERSION_UPGRADED_STRING);
+
         Iterator<String> iterator = settings.keys().iterator();
         while (iterator.hasNext()) {
             String key = iterator.next();

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

@@ -145,7 +145,11 @@ public void testValidation() throws IOException {
                 .put("index.analysis.analyzer.my_analyzer.type", "custom")
                 .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), customMetaData);
             Exception e = expectThrows(IllegalArgumentException.class, () -> validate(request, leaderIMD, followIMD, UUIDs, null));
-            assertThat(e.getMessage(), equalTo("the leader and follower index settings must be identical"));
+            assertThat(e.getMessage(), equalTo("the leader index setting[{\"index.analysis.analyzer.my_analyzer.tokenizer\"" +
+                ":\"whitespace\",\"index.analysis.analyzer.my_analyzer.type\":\"custom\",\"index.number_of_shards\":\"5\"," +
+                "\"index.soft_deletes.enabled\":\"true\"}] and follower index settings [{\"index.analysis.analyzer.my_analyzer.tokenizer" +
+                "\":\"standard\",\"index.analysis.analyzer.my_analyzer.type\":\"custom\",\"index.number_of_shards\":\"5\"," +
+                "\"index.soft_deletes.enabled\":\"true\"}] must be identical"));
         }
         {
             // should fail because the following index does not have the following_index settings
@@ -236,6 +240,20 @@ public void testDynamicIndexSettingsAreClassified() {
         }
     }
 
+    public void testFilter() {
+        Settings.Builder settings = Settings.builder();
+        settings.put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), "");
+        settings.put(IndexMetaData.SETTING_INDEX_VERSION_CREATED.getKey(), "");
+        settings.put(IndexMetaData.SETTING_INDEX_UUID, "");
+        settings.put(IndexMetaData.SETTING_INDEX_PROVIDED_NAME, "");
+        settings.put(IndexMetaData.SETTING_CREATION_DATE, "");
+        settings.put(IndexMetaData.SETTING_VERSION_UPGRADED, "");
+        settings.put(IndexMetaData.SETTING_VERSION_UPGRADED_STRING, "");
+
+        Settings result = TransportResumeFollowAction.filter(settings.build());
+        assertThat(result.size(), equalTo(0));
+    }
+
     private static IndexMetaData createIMD(String index,
                                            int numberOfShards,
                                            Settings settings,

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.tasks.TaskId;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -172,6 +173,12 @@ public Request(StreamInput in) throws IOException {
             super(in);
             this.remoteCluster = in.readString();
             this.leaderIndex = in.readString();
+            // Prior to 6.7.0 this request class used ResumeFollowAction.Request class to encapsulate followerIndex and
+            // followParameters fields. Now we need read and ignore fields that exist in super classes of ResumeFollowAction.Request
+            if (in.getVersion().before(Version.V_6_7_0)) {
+                TaskId.readFromStream(in); // TransportRequest.parentTaskId field
+                in.readTimeValue(); // MasterNodeRequest.masterNodeTimeout field
+            }
             this.followerIndex = in.readString();
             this.parameters = new FollowParameters(in);
             if (in.getVersion().onOrAfter(Version.V_6_7_0)) {
@@ -184,6 +191,12 @@ public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeString(remoteCluster);
             out.writeString(leaderIndex);
+            // Prior to 6.7.0 this request class used ResumeFollowAction.Request class to encapsulate followerIndex and
+            // followParameters fields. Now we need write fields that exist in super classes of ResumeFollowAction.Request
+            if (out.getVersion().before(Version.V_6_7_0)) {
+                TaskId.EMPTY_TASK_ID.writeTo(out); // TransportRequest.parentTaskId field
+                out.writeTimeValue(DEFAULT_MASTER_NODE_TIMEOUT); // MasterNodeRequest.masterNodeTimeout field
+            }
             out.writeString(followerIndex);
             parameters.writeTo(out);
             if (out.getVersion().onOrAfter(Version.V_6_7_0)) {