Map data tiers roles onto DATA legacy role for <7.3 (#71628)

Map data tiers roles onto DATA legacy role for <7.3

Author: andreidan

server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java

@@ -24,7 +24,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
@@ -339,8 +338,11 @@ public void writeTo(StreamOutput out) throws IOException {
             }
         } else {
             // an old node will only understand legacy roles since pluggable roles is a new concept
-            final List<DiscoveryNodeRole> rolesToWrite =
-                    roles.stream().filter(DiscoveryNodeRole.LEGACY_ROLES::contains).collect(Collectors.toList());
+            final Set<DiscoveryNodeRole> rolesToWrite = roles.stream()
+                .map(role -> role.getCompatibilityRole(out.getVersion()))
+                .filter(DiscoveryNodeRole.LEGACY_ROLES::contains)
+                .collect(Collectors.toSet());
+
             out.writeVInt(rolesToWrite.size());
             for (final DiscoveryNodeRole role : rolesToWrite) {
                 if (role == DiscoveryNodeRole.MASTER_ROLE) {

server/src/test/java/org/elasticsearch/cluster/node/DiscoveryNodeTests.java

@@ -137,6 +137,22 @@ public DiscoveryNodeRole getCompatibilityRole(Version nodeVersion) {
                 equalTo("data"));
         }
 
+        {
+            // a pre 7.3.0 node will only understand legacy roles so let's test a custom data containing node role is mapped onto the
+            // `DATA` role
+            DiscoveryNode nodeToWrite = new DiscoveryNode("name1", "id1", transportAddress, emptyMap(),
+                org.elasticsearch.common.collect.Set.of(customRole, DiscoveryNodeRole.MASTER_ROLE), Version.CURRENT);
+
+            BytesStreamOutput streamOutput = new BytesStreamOutput();
+            streamOutput.setVersion(Version.V_7_2_0);
+            nodeToWrite.writeTo(streamOutput);
+
+            StreamInput in = StreamInput.wrap(streamOutput.bytes().toBytesRef().bytes);
+            in.setVersion(Version.V_7_2_0);
+            DiscoveryNode serialized = new DiscoveryNode(in);
+            assertThat(serialized.getRoles().stream().map(DiscoveryNodeRole::roleName).sorted().collect(Collectors.joining(",")),
+                equalTo("data,master"));
+        }
     }
 
     public void testDiscoveryNodeIsRemoteClusterClientDefault() {