diff --git a/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java b/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java
index 94bddd40bd919..02563f095eec4 100644
--- a/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java
+++ b/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java
@@ -325,8 +325,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_7_3_0)) {
             out.writeVInt(roles.size());
             for (final DiscoveryNodeRole role : roles) {
-                out.writeString(role.roleName());
-                out.writeString(role.roleNameAbbreviation());
+                final DiscoveryNodeRole compatibleRole = role.getCompatibilityRole(out.getVersion());
+                out.writeString(compatibleRole.roleName());
+                out.writeString(compatibleRole.roleNameAbbreviation());
             }
         } else {
             // an old node will only understand legacy roles since pluggable roles is a new concept
diff --git a/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeRole.java b/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeRole.java
index d72a7042d7091..92667749d7666 100644
--- a/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeRole.java
+++ b/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeRole.java
@@ -91,6 +91,16 @@ public boolean canContainData() {
         return false;
     }
 
+    /**
+     * When serializing a {@link DiscoveryNodeRole}, the role may not be available to nodes of
+     * previous versions, where the role had not yet been added. This method allows overriding
+     * the role that should be serialized when communicating to versions prior to the introduction
+     * of the discovery node role.
+     */
+    public DiscoveryNodeRole getCompatibilityRole(Version nodeVersion) {
+        return this;
+    }
+
     @Override
     public final boolean equals(Object o) {
         if (this == o) return true;
diff --git a/server/src/test/java/org/elasticsearch/cluster/node/DiscoveryNodeTests.java b/server/src/test/java/org/elasticsearch/cluster/node/DiscoveryNodeTests.java
index 38168ef5e2ee7..8a049093b82f5 100644
--- a/server/src/test/java/org/elasticsearch/cluster/node/DiscoveryNodeTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/node/DiscoveryNodeTests.java
@@ -22,13 +22,16 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.test.ESTestCase;
 
 import java.net.InetAddress;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.emptySet;
@@ -88,6 +91,53 @@ public void testDiscoveryNodeSerializationKeepsHost() throws Exception {
         assertEquals(transportAddress.getPort(), serialized.getAddress().getPort());
     }
 
+    public void testDiscoveryNodeRoleWithOldVersion() throws Exception {
+        InetAddress inetAddress = InetAddress.getByAddress("name1", new byte[] { (byte) 192, (byte) 168, (byte) 0, (byte) 1});
+        TransportAddress transportAddress = new TransportAddress(inetAddress, randomIntBetween(0, 65535));
+
+        DiscoveryNodeRole customRole = new DiscoveryNodeRole("custom_role", "z") {
+            @Override
+            public Setting<Boolean> legacySetting() {
+                return null;
+            }
+
+            @Override
+            public DiscoveryNodeRole getCompatibilityRole(Version nodeVersion) {
+                if (nodeVersion.equals(Version.CURRENT)) {
+                    return this;
+                } else {
+                    return DiscoveryNodeRole.DATA_ROLE;
+                }
+            }
+        };
+
+        DiscoveryNode node = new DiscoveryNode("name1", "id1", transportAddress, emptyMap(),
+            Collections.singleton(customRole), Version.CURRENT);
+
+        {
+            BytesStreamOutput streamOutput = new BytesStreamOutput();
+            streamOutput.setVersion(Version.CURRENT);
+            node.writeTo(streamOutput);
+
+            StreamInput in = StreamInput.wrap(streamOutput.bytes().toBytesRef().bytes);
+            DiscoveryNode serialized = new DiscoveryNode(in);
+            assertThat(serialized.getRoles().stream().map(DiscoveryNodeRole::roleName).collect(Collectors.joining()),
+                equalTo("custom_role"));
+        }
+
+        {
+            BytesStreamOutput streamOutput = new BytesStreamOutput();
+            streamOutput.setVersion(Version.V_7_10_0);
+            node.writeTo(streamOutput);
+
+            StreamInput in = StreamInput.wrap(streamOutput.bytes().toBytesRef().bytes);
+            DiscoveryNode serialized = new DiscoveryNode(in);
+            assertThat(serialized.getRoles().stream().map(DiscoveryNodeRole::roleName).collect(Collectors.joining()),
+                equalTo("data"));
+        }
+
+    }
+
     public void testDiscoveryNodeIsRemoteClusterClientDefault() {
         runTestDiscoveryNodeIsRemoteClusterClient(Settings.EMPTY, true);
     }
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTier.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTier.java
index 02728b08312f4..bc12b8f04315a 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTier.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTier.java
@@ -8,6 +8,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
@@ -80,6 +81,11 @@ public Setting<Boolean> legacySetting() {
         public boolean canContainData() {
             return true;
         }
+
+        @Override
+        public DiscoveryNodeRole getCompatibilityRole(Version nodeVersion) {
+            return nodeVersion.before(Version.V_7_10_0) ? DiscoveryNodeRole.DATA_ROLE : this;
+        }
     };
 
     public static DiscoveryNodeRole DATA_HOT_NODE_ROLE = new DiscoveryNodeRole("data_hot", "h") {
@@ -97,6 +103,11 @@ public Setting<Boolean> legacySetting() {
         public boolean canContainData() {
             return true;
         }
+
+        @Override
+        public DiscoveryNodeRole getCompatibilityRole(Version nodeVersion) {
+            return nodeVersion.before(Version.V_7_10_0) ? DiscoveryNodeRole.DATA_ROLE : this;
+        }
     };
 
     public static DiscoveryNodeRole DATA_WARM_NODE_ROLE = new DiscoveryNodeRole("data_warm", "w") {
@@ -114,6 +125,11 @@ public Setting<Boolean> legacySetting() {
         public boolean canContainData() {
             return true;
         }
+
+        @Override
+        public DiscoveryNodeRole getCompatibilityRole(Version nodeVersion) {
+            return nodeVersion.before(Version.V_7_10_0) ? DiscoveryNodeRole.DATA_ROLE : this;
+        }
     };
 
     public static DiscoveryNodeRole DATA_COLD_NODE_ROLE = new DiscoveryNodeRole("data_cold", "c") {
@@ -131,6 +147,11 @@ public Setting<Boolean> legacySetting() {
         public boolean canContainData() {
             return true;
         }
+
+        @Override
+        public DiscoveryNodeRole getCompatibilityRole(Version nodeVersion) {
+            return nodeVersion.before(Version.V_7_10_0) ? DiscoveryNodeRole.DATA_ROLE : this;
+        }
     };
 
     public static boolean isContentNode(DiscoveryNode discoveryNode) {
diff --git a/x-pack/qa/mixed-tier-cluster/build.gradle b/x-pack/qa/mixed-tier-cluster/build.gradle
new file mode 100644
index 0000000000000..9613a1af7c0eb
--- /dev/null
+++ b/x-pack/qa/mixed-tier-cluster/build.gradle
@@ -0,0 +1,61 @@
+apply plugin: 'elasticsearch.testclusters'
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply from : "$rootDir/gradle/bwc-test.gradle"
+apply plugin: 'elasticsearch.rest-test'
+
+import org.elasticsearch.gradle.Version
+import org.elasticsearch.gradle.VersionProperties
+import org.elasticsearch.gradle.info.BuildParams
+import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask
+
+dependencies {
+  testImplementation project(':x-pack:qa')
+}
+
+// Only run tests for 7.9+, since the node.roles setting was introduced in 7.9.0
+for (Version bwcVersion : BuildParams.bwcVersions.wireCompatible.findAll { it.onOrAfter('7.9.0') }) {
+  if (bwcVersion == VersionProperties.getElasticsearchVersion()) {
+    // Not really a mixed cluster
+    continue;
+  }
+
+  String baseName = "v${bwcVersion}"
+
+  testClusters {
+    "${baseName}" {
+      versions = [bwcVersion.toString(), project.version]
+      numberOfNodes = 3
+      testDistribution = 'DEFAULT'
+      setting 'xpack.security.enabled', 'false'
+      setting 'xpack.watcher.enabled', 'false'
+      setting 'xpack.ml.enabled', 'false'
+      setting 'xpack.license.self_generated.type', 'trial'
+      nodes."${baseName}-0".setting 'node.roles', '["master"]'
+      // data_* roles were introduced in 7.10.0, so use 'data' for older versions
+      if (bwcVersion.before('7.10.0')) {
+        nodes."${baseName}-1".setting 'node.roles', '["data"]'
+      } else {
+        nodes."${baseName}-1".setting 'node.roles', '["data_content", "data_hot"]'
+      }
+      nodes."${baseName}-2".setting 'node.roles', '["master"]'
+    }
+  }
+
+  tasks.register("${baseName}#mixedClusterTest", StandaloneRestIntegTestTask) {
+    useCluster testClusters."${baseName}"
+    mustRunAfter(precommit)
+    doFirst {
+      // Getting the endpoints causes a wait for the cluster
+      println "Endpoints are: ${-> testClusters."${baseName}".allHttpSocketURI.join(",")}"
+      testClusters."${baseName}".nextNodeToNextVersion()
+
+      nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}".allHttpSocketURI.join(",")}")
+      nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}".getName()}")
+    }
+    onlyIf { project.bwc_tests_enabled }
+  }
+
+  tasks.register(bwcTaskName(bwcVersion)) {
+    dependsOn "${baseName}#mixedClusterTest"
+  }
+}
diff --git a/x-pack/qa/mixed-tier-cluster/src/test/java/org/elasticsearch/mixed/DataTierMixedIT.java b/x-pack/qa/mixed-tier-cluster/src/test/java/org/elasticsearch/mixed/DataTierMixedIT.java
new file mode 100644
index 0000000000000..303fe3ee4b238
--- /dev/null
+++ b/x-pack/qa/mixed-tier-cluster/src/test/java/org/elasticsearch/mixed/DataTierMixedIT.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.mixed;
+
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.rest.ESRestTestCase;
+
+public class DataTierMixedIT extends ESRestTestCase {
+
+    public void testMixedTierCompatibility() throws Exception {
+        createIndex("test-index", Settings.builder()
+            .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+            .build());
+        ensureGreen("test-index");
+    }
+}