From 42223eb71adc5764a0fec91a95dee1702f6ef4ac Mon Sep 17 00:00:00 2001
From: Patrik Nordwall <patrik.nordwall@gmail.com>
Date: Thu, 10 Sep 2020 10:42:03 +0200
Subject: [PATCH] Add app-version to the Member information, #27300 (#29546)

* will be used in rolling update features
* configured with akka.cluster.app-version
* reusing same implementation as ManifestInfo.Version
  by moving that to akka.util.Version
* additional version test
* support dynver format, + separator, and commit number
* improve version parser
* lazy parse
* make Member.appVersion internal
---
 .../scala/akka/util/ManifestInfoSpec.scala    |   9 +-
 .../akka/util/ManifestInfoVersionSpec.scala   |  39 --
 .../test/scala/akka/util/VersionSpec.scala    |  95 +++
 .../main/scala/akka/util/ManifestInfo.scala   |  57 +-
 .../src/main/scala/akka/util/Version.scala    | 192 ++++++
 .../cluster/protobuf/msg/ClusterMessages.java | 610 ++++++++++++++++--
 .../issue-27300-appVersion.excludes           |   9 +
 .../src/main/protobuf/ClusterMessages.proto   |   3 +
 .../src/main/resources/reference.conf         |  17 +
 .../scala/akka/cluster/ClusterDaemon.scala    |  26 +-
 .../scala/akka/cluster/ClusterEvent.scala     |  13 +
 .../main/scala/akka/cluster/ClusterJmx.scala  |   7 +-
 .../scala/akka/cluster/ClusterReadView.scala  |   3 +-
 .../scala/akka/cluster/ClusterSettings.scala  |   4 +
 .../src/main/scala/akka/cluster/Member.scala  |  30 +-
 .../protobuf/ClusterMessageSerializer.scala   |  35 +-
 .../scala/akka/cluster/JoinSeedNodeSpec.scala |  19 +-
 .../scala/akka/cluster/MBeanSpec.scala        |  13 +-
 .../akka/cluster/ClusterConfigSpec.scala      |   2 +
 .../cluster/ClusterHeartbeatSenderSpec.scala  |   4 +-
 .../test/scala/akka/cluster/ClusterSpec.scala |   7 +-
 .../cluster/CrossDcHeartbeatSenderSpec.scala  |   4 +-
 .../akka/cluster/MemberOrderingSpec.scala     |   8 +-
 .../test/scala/akka/cluster/TestMember.scala  |  11 +-
 .../ClusterMessageSerializerSpec.scala        |  30 +-
 .../akka/cluster/sbr/TestAddresses.scala      |  38 +-
 26 files changed, 1067 insertions(+), 218 deletions(-)
 delete mode 100644 akka-actor-tests/src/test/scala/akka/util/ManifestInfoVersionSpec.scala
 create mode 100644 akka-actor-tests/src/test/scala/akka/util/VersionSpec.scala
 create mode 100644 akka-actor/src/main/scala/akka/util/Version.scala
 create mode 100644 akka-cluster/src/main/mima-filters/2.6.8.backwards.excludes/issue-27300-appVersion.excludes

diff --git a/akka-actor-tests/src/test/scala/akka/util/ManifestInfoSpec.scala b/akka-actor-tests/src/test/scala/akka/util/ManifestInfoSpec.scala
index 9fcf883dcc..d38a035970 100644
--- a/akka-actor-tests/src/test/scala/akka/util/ManifestInfoSpec.scala
+++ b/akka-actor-tests/src/test/scala/akka/util/ManifestInfoSpec.scala
@@ -5,16 +5,15 @@
 package akka.util
 
 import akka.testkit.AkkaSpec
-import akka.util.ManifestInfo.Version
 
 class ManifestInfoSpec extends AkkaSpec {
   "ManifestInfo" should {
     "produce a clear message" in {
       val versions = Map(
-        "akka-actor" -> new Version("2.6.4"),
-        "akka-persistence" -> new Version("2.5.3"),
-        "akka-cluster" -> new Version("2.5.3"),
-        "unrelated" -> new Version("2.5.3"))
+        "akka-actor" -> new ManifestInfo.Version("2.6.4"),
+        "akka-persistence" -> new ManifestInfo.Version("2.5.3"),
+        "akka-cluster" -> new ManifestInfo.Version("2.5.3"),
+        "unrelated" -> new ManifestInfo.Version("2.5.3"))
       val allModules = List("akka-actor", "akka-persistence", "akka-cluster")
       ManifestInfo.checkSameVersion("Akka", allModules, versions) shouldBe Some(
         "You are using version 2.6.4 of Akka, but it appears you (perhaps indirectly) also depend on older versions of related artifacts. " +
diff --git a/akka-actor-tests/src/test/scala/akka/util/ManifestInfoVersionSpec.scala b/akka-actor-tests/src/test/scala/akka/util/ManifestInfoVersionSpec.scala
deleted file mode 100644
index 19aab3c4dc..0000000000
--- a/akka-actor-tests/src/test/scala/akka/util/ManifestInfoVersionSpec.scala
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Copyright (C) 2015-2020 Lightbend Inc. <https://www.lightbend.com>
- */
-
-package akka.util
-
-import org.scalatest.matchers.should.Matchers
-import org.scalatest.wordspec.AnyWordSpec
-
-import akka.util.ManifestInfo.Version
-
-class ManifestInfoVersionSpec extends AnyWordSpec with Matchers {
-
-  "Version" should {
-
-    "compare full version" in {
-      new Version("1.2.3") should ===(new Version("1.2.3"))
-      new Version("1.2.3") should !==(new Version("1.2.4"))
-      new Version("1.2.4") should be > new Version("1.2.3")
-      new Version("3.2.1") should be > new Version("1.2.3")
-      new Version("3.2.1") should be < new Version("3.3.1")
-    }
-
-    "compare partial version" in {
-      new Version("1.2") should ===(new Version("1.2"))
-      new Version("1.2") should !==(new Version("1.3"))
-      new Version("1.2.1") should be > new Version("1.2")
-      new Version("2.4") should be > new Version("2.3")
-      new Version("3.2") should be < new Version("3.2.7")
-    }
-
-    "compare extra" in {
-      new Version("1.2.3-foo") should ===(new Version("1.2.3-foo"))
-      new Version("1.2.3-foo") should !==(new Version("1.2.3-bar"))
-      new Version("1.2-foo") should be > new Version("1.2.3")
-      new Version("1.2.3-foo") should be > new Version("1.2.3-bar")
-    }
-  }
-}
diff --git a/akka-actor-tests/src/test/scala/akka/util/VersionSpec.scala b/akka-actor-tests/src/test/scala/akka/util/VersionSpec.scala
new file mode 100644
index 0000000000..636a542d65
--- /dev/null
+++ b/akka-actor-tests/src/test/scala/akka/util/VersionSpec.scala
@@ -0,0 +1,95 @@
+/*
+ * Copyright (C) 2015-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.util
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class VersionSpec extends AnyWordSpec with Matchers {
+
+  "Version" should {
+
+    "compare 3 digit version" in {
+      Version("1.2.3") should ===(Version("1.2.3"))
+      Version("1.2.3") should !==(Version("1.2.4"))
+      Version("1.2.4") should be > Version("1.2.3")
+      Version("3.2.1") should be > Version("1.2.3")
+      Version("3.2.1") should be < Version("3.3.1")
+      Version("3.2.0") should be < Version("3.2.1")
+    }
+
+    "not support more than 3 digits version" in {
+      intercept[IllegalArgumentException](Version("1.2.3.1"))
+    }
+
+    "compare 2 digit version" in {
+      Version("1.2") should ===(Version("1.2"))
+      Version("1.2") should ===(Version("1.2.0"))
+      Version("1.2") should !==(Version("1.3"))
+      Version("1.2.1") should be > Version("1.2")
+      Version("2.4") should be > Version("2.3")
+      Version("3.2") should be < Version("3.2.7")
+    }
+
+    "compare single digit version" in {
+      Version("1") should ===(Version("1"))
+      Version("1") should ===(Version("1.0"))
+      Version("1") should ===(Version("1.0.0"))
+      Version("1") should !==(Version("2"))
+      Version("3") should be > Version("2")
+
+      Version("2b") should be > Version("2a")
+      Version("2020-09-07") should be > Version("2020-08-30")
+    }
+
+    "compare extra" in {
+      Version("1.2.3-M1") should ===(Version("1.2.3-M1"))
+      Version("1.2-M1") should ===(Version("1.2-M1"))
+      Version("1.2.0-M1") should ===(Version("1.2-M1"))
+      Version("1.2.3-M1") should !==(Version("1.2.3-M2"))
+      Version("1.2-M1") should be < Version("1.2.0")
+      Version("1.2.0-M1") should be < Version("1.2.0")
+      Version("1.2.3-M2") should be > Version("1.2.3-M1")
+    }
+
+    "require digits" in {
+      intercept[NumberFormatException](Version("1.x.3"))
+      intercept[NumberFormatException](Version("1.2x.3"))
+      intercept[NumberFormatException](Version("1.2.x"))
+      intercept[NumberFormatException](Version("1.2.3x"))
+
+      intercept[NumberFormatException](Version("x.3"))
+      intercept[NumberFormatException](Version("1.x"))
+      intercept[NumberFormatException](Version("1.2x"))
+    }
+
+    "compare dynver format" in {
+      // dynver format
+      Version("1.0.10+3-1234abcd") should be < Version("1.0.11")
+      Version("1.0.10+3-1234abcd") should be < Version("1.0.10+10-1234abcd")
+      Version("1.2+3-1234abcd") should be < Version("1.2+10-1234abcd")
+      Version("1.0.0+3-1234abcd+20140707-1030") should be < Version("1.0.0+3-1234abcd+20140707-1130")
+      Version("0.0.0+3-2234abcd") should be < Version("0.0.0+4-1234abcd")
+      Version("HEAD+20140707-1030") should be < Version("HEAD+20140707-1130")
+
+      Version("1.0.10-3-1234abcd") should be < Version("1.0.10-10-1234abcd")
+      Version("1.0.0-3-1234abcd+20140707-1030") should be < Version("1.0.0-3-1234abcd+20140707-1130")
+
+      // not real dynver, but should still work
+      Version("1.0.10+3a-1234abcd") should be < Version("1.0.10+3b-1234abcd")
+
+    }
+
+    "compare extra without digits" in {
+      Version("foo") should ===(Version("foo"))
+      Version("foo") should !==(Version("bar"))
+      Version("foo") should be < Version("1.2.3")
+      Version("foo") should be > Version("bar")
+
+      Version("1-foo") should !==(Version("01-foo"))
+      Version("1-foo") should be > (Version("02-foo"))
+    }
+  }
+}
diff --git a/akka-actor/src/main/scala/akka/util/ManifestInfo.scala b/akka-actor/src/main/scala/akka/util/ManifestInfo.scala
index 913050be75..7bca50758b 100644
--- a/akka-actor/src/main/scala/akka/util/ManifestInfo.scala
+++ b/akka-actor/src/main/scala/akka/util/ManifestInfo.scala
@@ -5,7 +5,6 @@
 package akka.util
 
 import java.io.IOException
-import java.util.Arrays
 import java.util.jar.Attributes
 import java.util.jar.Manifest
 
@@ -53,61 +52,21 @@ object ManifestInfo extends ExtensionId[ManifestInfo] with ExtensionIdProvider {
    * Comparable version information
    */
   final class Version(val version: String) extends Comparable[Version] {
-    private val (numbers: Array[Int], rest: String) = {
-      val numbers = new Array[Int](3)
-      val segments: Array[String] = version.split("[.-]")
-      var segmentPos = 0
-      var numbersPos = 0
-      while (numbersPos < 3) {
-        if (segmentPos < segments.length) try {
-          numbers(numbersPos) = segments(segmentPos).toInt
-          segmentPos += 1
-        } catch {
-          case _: NumberFormatException =>
-            // This means that we have a trailing part on the version string and
-            // less than 3 numbers, so we assume that this is a "newer" version
-            numbers(numbersPos) = Integer.MAX_VALUE
-        }
-        numbersPos += 1
-      }
+    private val impl = new akka.util.Version(version)
 
-      val rest: String =
-        if (segmentPos >= segments.length) ""
-        else String.join("-", Arrays.asList(Arrays.copyOfRange(segments, segmentPos, segments.length): _*))
-
-      (numbers, rest)
-    }
-
-    override def compareTo(other: Version): Int = {
-      var diff = 0
-      diff = numbers(0) - other.numbers(0)
-      if (diff == 0) {
-        diff = numbers(1) - other.numbers(1)
-        if (diff == 0) {
-          diff = numbers(2) - other.numbers(2)
-          if (diff == 0) {
-            diff = rest.compareTo(other.rest)
-          }
-        }
-      }
-      diff
-    }
+    override def compareTo(other: Version): Int =
+      impl.compareTo(other.impl)
 
     override def equals(o: Any): Boolean = o match {
-      case v: Version => compareTo(v) == 0
+      case v: Version => impl.equals(v.impl)
       case _          => false
     }
 
-    override def hashCode(): Int = {
-      var result = HashCode.SEED
-      result = HashCode.hash(result, numbers(0))
-      result = HashCode.hash(result, numbers(1))
-      result = HashCode.hash(result, numbers(2))
-      result = HashCode.hash(result, rest)
-      result
-    }
+    override def hashCode(): Int =
+      impl.hashCode()
 
-    override def toString: String = version
+    override def toString: String =
+      impl.toString
   }
 
   /** INTERNAL API */
diff --git a/akka-actor/src/main/scala/akka/util/Version.scala b/akka-actor/src/main/scala/akka/util/Version.scala
new file mode 100644
index 0000000000..a5fdba142f
--- /dev/null
+++ b/akka-actor/src/main/scala/akka/util/Version.scala
@@ -0,0 +1,192 @@
+/*
+ * Copyright (C) 2015-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.util
+
+import akka.annotation.InternalApi
+
+object Version {
+  val Zero: Version = Version("0.0.0")
+
+  private val Undefined = 0
+
+  def apply(version: String): Version = {
+    val v = new Version(version)
+    v.parse()
+  }
+}
+
+/**
+ * Comparable version information.
+ *
+ * The typical convention is to use 3 digit version numbers `major.minor.patch`,
+ * but 1 or two digits are also supported.
+ *
+ * If no `.` is used it is interpreted as a single digit version number or as
+ * plain alphanumeric if it couldn't be parsed as a number.
+ *
+ * It may also have a qualifier at the end for 2 or 3 digit version numbers such as "1.2-RC1".
+ * For 1 digit with qualifier, 1-RC1, it is interpreted as plain alphanumeric.
+ *
+ * It has support for https://github.com/dwijnand/sbt-dynver format with `+` or
+ * `-` separator. The number of commits from the tag is handled as a numeric part.
+ * For example `1.0.0+3-73475dce26` is less than `1.0.10+10-ed316bd024` (3 < 10).
+ */
+final class Version(val version: String) extends Comparable[Version] {
+  import Version.Undefined
+
+  // lazy initialized via `parse` method
+  @volatile private var numbers: Array[Int] = Array.emptyIntArray
+  private var rest: String = ""
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] def parse(): Version = {
+    def parseLastPart(s: String): (Int, String) = {
+      // for example 2, 2-SNAPSHOT or dynver 2+10-1234abcd
+      if (s.length == 0) {
+        Undefined -> s
+      } else {
+        val i = s.indexOf('-')
+        val j = s.indexOf('+') // for dynver
+        val k =
+          if (i == -1) j
+          else if (j == -1) i
+          else math.min(i, j)
+        if (k == -1)
+          s.toInt -> ""
+        else
+          s.substring(0, k).toInt -> s.substring(k + 1)
+      }
+    }
+
+    def parseDynverPart(s: String): (Int, String) = {
+      // for example SNAPSHOT or dynver 10-1234abcd
+      if (s.isEmpty || !s.charAt(0).isDigit) {
+        Undefined -> s
+      } else {
+        s.indexOf('-') match {
+          case -1 =>
+            Undefined -> s
+          case i =>
+            try {
+              s.substring(0, i).toInt -> s.substring(i + 1)
+            } catch {
+              case _: NumberFormatException =>
+                Undefined -> s
+            }
+        }
+      }
+    }
+
+    def parseLastParts(s: String): (Int, Int, String) = {
+      // for example 2, 2-SNAPSHOT or dynver 2+10-1234abcd
+      val (lastNumber, rest) = parseLastPart(s)
+      if (rest == "")
+        (lastNumber, Undefined, rest)
+      else {
+        val (dynverNumber, rest2) = parseDynverPart(rest)
+        (lastNumber, dynverNumber, rest2)
+      }
+    }
+
+    if (numbers.length == 0) {
+
+      val nbrs = new Array[Int](4)
+      val segments = version.split('.')
+
+      val rst =
+        if (segments.length == 1) {
+          // single digit or alphanumeric
+          val s = segments(0)
+          if (s.isEmpty)
+            throw new IllegalArgumentException("Empty version not supported.")
+          nbrs(1) = Undefined
+          nbrs(2) = Undefined
+          nbrs(3) = Undefined
+          if (s.charAt(0).isDigit) {
+            try {
+              nbrs(0) = s.toInt
+              ""
+            } catch {
+              case _: NumberFormatException =>
+                s
+            }
+          } else {
+            s
+          }
+        } else if (segments.length == 2) {
+          // for example 1.2, 1.2-SNAPSHOT or dynver 1.2+10-1234abcd
+          val (n1, n2, rest) = parseLastParts(segments(1))
+          nbrs(0) = segments(0).toInt
+          nbrs(1) = n1
+          nbrs(2) = n2
+          nbrs(3) = Undefined
+          rest
+        } else if (segments.length == 3) {
+          // for example 1.2.3, 1.2.3-SNAPSHOT or dynver 1.2.3+10-1234abcd
+          val (n1, n2, rest) = parseLastParts(segments(2))
+          nbrs(0) = segments(0).toInt
+          nbrs(1) = segments(1).toInt
+          nbrs(2) = n1
+          nbrs(3) = n2
+          rest
+        } else {
+          throw new IllegalArgumentException(s"Only 3 digits separated with '.' are supported. [$version]")
+        }
+
+      this.rest = rst
+      this.numbers = nbrs
+    }
+    this
+  }
+
+  override def compareTo(other: Version): Int = {
+    if (version == other.version) // String equals without requiring parse
+      0
+    else {
+      parse()
+      other.parse()
+      var diff = 0
+      diff = numbers(0) - other.numbers(0)
+      if (diff == 0) {
+        diff = numbers(1) - other.numbers(1)
+        if (diff == 0) {
+          diff = numbers(2) - other.numbers(2)
+          if (diff == 0) {
+            diff = numbers(3) - other.numbers(3)
+            if (diff == 0) {
+              if (rest == "" && other.rest != "")
+                diff = 1
+              if (other.rest == "" && rest != "")
+                diff = -1
+              else
+                diff = rest.compareTo(other.rest)
+            }
+          }
+        }
+      }
+      diff
+    }
+  }
+
+  override def equals(o: Any): Boolean = o match {
+    case v: Version => compareTo(v) == 0
+    case _          => false
+  }
+
+  override def hashCode(): Int = {
+    parse()
+    var result = HashCode.SEED
+    result = HashCode.hash(result, numbers(0))
+    result = HashCode.hash(result, numbers(1))
+    result = HashCode.hash(result, numbers(2))
+    result = HashCode.hash(result, numbers(3))
+    result = HashCode.hash(result, rest)
+    result
+  }
+
+  override def toString: String = version
+}
diff --git a/akka-cluster/src/main/java/akka/cluster/protobuf/msg/ClusterMessages.java b/akka-cluster/src/main/java/akka/cluster/protobuf/msg/ClusterMessages.java
index 3c7b093856..f22a5441b3 100644
--- a/akka-cluster/src/main/java/akka/cluster/protobuf/msg/ClusterMessages.java
+++ b/akka-cluster/src/main/java/akka/cluster/protobuf/msg/ClusterMessages.java
@@ -317,6 +317,23 @@ public final class ClusterMessages {
      */
     akka.protobufv3.internal.ByteString
         getRolesBytes(int index);
+
+    /**
+     * <code>optional string appVersion = 3;</code>
+     * @return Whether the appVersion field is set.
+     */
+    boolean hasAppVersion();
+    /**
+     * <code>optional string appVersion = 3;</code>
+     * @return The appVersion.
+     */
+    java.lang.String getAppVersion();
+    /**
+     * <code>optional string appVersion = 3;</code>
+     * @return The bytes for appVersion.
+     */
+    akka.protobufv3.internal.ByteString
+        getAppVersionBytes();
   }
   /**
    * <pre>
@@ -337,6 +354,7 @@ public final class ClusterMessages {
     }
     private Join() {
       roles_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
+      appVersion_ = "";
     }
 
     @java.lang.Override
@@ -392,6 +410,12 @@ public final class ClusterMessages {
               roles_.add(bs);
               break;
             }
+            case 26: {
+              akka.protobufv3.internal.ByteString bs = input.readBytes();
+              bitField0_ |= 0x00000002;
+              appVersion_ = bs;
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -486,6 +510,51 @@ public final class ClusterMessages {
       return roles_.getByteString(index);
     }
 
+    public static final int APPVERSION_FIELD_NUMBER = 3;
+    private volatile java.lang.Object appVersion_;
+    /**
+     * <code>optional string appVersion = 3;</code>
+     * @return Whether the appVersion field is set.
+     */
+    public boolean hasAppVersion() {
+      return ((bitField0_ & 0x00000002) != 0);
+    }
+    /**
+     * <code>optional string appVersion = 3;</code>
+     * @return The appVersion.
+     */
+    public java.lang.String getAppVersion() {
+      java.lang.Object ref = appVersion_;
+      if (ref instanceof java.lang.String) {
+        return (java.lang.String) ref;
+      } else {
+        akka.protobufv3.internal.ByteString bs = 
+            (akka.protobufv3.internal.ByteString) ref;
+        java.lang.String s = bs.toStringUtf8();
+        if (bs.isValidUtf8()) {
+          appVersion_ = s;
+        }
+        return s;
+      }
+    }
+    /**
+     * <code>optional string appVersion = 3;</code>
+     * @return The bytes for appVersion.
+     */
+    public akka.protobufv3.internal.ByteString
+        getAppVersionBytes() {
+      java.lang.Object ref = appVersion_;
+      if (ref instanceof java.lang.String) {
+        akka.protobufv3.internal.ByteString b = 
+            akka.protobufv3.internal.ByteString.copyFromUtf8(
+                (java.lang.String) ref);
+        appVersion_ = b;
+        return b;
+      } else {
+        return (akka.protobufv3.internal.ByteString) ref;
+      }
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -514,6 +583,9 @@ public final class ClusterMessages {
       for (int i = 0; i < roles_.size(); i++) {
         akka.protobufv3.internal.GeneratedMessageV3.writeString(output, 2, roles_.getRaw(i));
       }
+      if (((bitField0_ & 0x00000002) != 0)) {
+        akka.protobufv3.internal.GeneratedMessageV3.writeString(output, 3, appVersion_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -535,6 +607,9 @@ public final class ClusterMessages {
         size += dataSize;
         size += 1 * getRolesList().size();
       }
+      if (((bitField0_ & 0x00000002) != 0)) {
+        size += akka.protobufv3.internal.GeneratedMessageV3.computeStringSize(3, appVersion_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -557,6 +632,11 @@ public final class ClusterMessages {
       }
       if (!getRolesList()
           .equals(other.getRolesList())) return false;
+      if (hasAppVersion() != other.hasAppVersion()) return false;
+      if (hasAppVersion()) {
+        if (!getAppVersion()
+            .equals(other.getAppVersion())) return false;
+      }
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -576,6 +656,10 @@ public final class ClusterMessages {
         hash = (37 * hash) + ROLES_FIELD_NUMBER;
         hash = (53 * hash) + getRolesList().hashCode();
       }
+      if (hasAppVersion()) {
+        hash = (37 * hash) + APPVERSION_FIELD_NUMBER;
+        hash = (53 * hash) + getAppVersion().hashCode();
+      }
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -723,6 +807,8 @@ public final class ClusterMessages {
         bitField0_ = (bitField0_ & ~0x00000001);
         roles_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
         bitField0_ = (bitField0_ & ~0x00000002);
+        appVersion_ = "";
+        bitField0_ = (bitField0_ & ~0x00000004);
         return this;
       }
 
@@ -764,6 +850,10 @@ public final class ClusterMessages {
           bitField0_ = (bitField0_ & ~0x00000002);
         }
         result.roles_ = roles_;
+        if (((from_bitField0_ & 0x00000004) != 0)) {
+          to_bitField0_ |= 0x00000002;
+        }
+        result.appVersion_ = appVersion_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -826,6 +916,11 @@ public final class ClusterMessages {
           }
           onChanged();
         }
+        if (other.hasAppVersion()) {
+          bitField0_ |= 0x00000004;
+          appVersion_ = other.appVersion_;
+          onChanged();
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -1090,6 +1185,90 @@ public final class ClusterMessages {
         onChanged();
         return this;
       }
+
+      private java.lang.Object appVersion_ = "";
+      /**
+       * <code>optional string appVersion = 3;</code>
+       * @return Whether the appVersion field is set.
+       */
+      public boolean hasAppVersion() {
+        return ((bitField0_ & 0x00000004) != 0);
+      }
+      /**
+       * <code>optional string appVersion = 3;</code>
+       * @return The appVersion.
+       */
+      public java.lang.String getAppVersion() {
+        java.lang.Object ref = appVersion_;
+        if (!(ref instanceof java.lang.String)) {
+          akka.protobufv3.internal.ByteString bs =
+              (akka.protobufv3.internal.ByteString) ref;
+          java.lang.String s = bs.toStringUtf8();
+          if (bs.isValidUtf8()) {
+            appVersion_ = s;
+          }
+          return s;
+        } else {
+          return (java.lang.String) ref;
+        }
+      }
+      /**
+       * <code>optional string appVersion = 3;</code>
+       * @return The bytes for appVersion.
+       */
+      public akka.protobufv3.internal.ByteString
+          getAppVersionBytes() {
+        java.lang.Object ref = appVersion_;
+        if (ref instanceof String) {
+          akka.protobufv3.internal.ByteString b = 
+              akka.protobufv3.internal.ByteString.copyFromUtf8(
+                  (java.lang.String) ref);
+          appVersion_ = b;
+          return b;
+        } else {
+          return (akka.protobufv3.internal.ByteString) ref;
+        }
+      }
+      /**
+       * <code>optional string appVersion = 3;</code>
+       * @param value The appVersion to set.
+       * @return This builder for chaining.
+       */
+      public Builder setAppVersion(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  bitField0_ |= 0x00000004;
+        appVersion_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional string appVersion = 3;</code>
+       * @return This builder for chaining.
+       */
+      public Builder clearAppVersion() {
+        bitField0_ = (bitField0_ & ~0x00000004);
+        appVersion_ = getDefaultInstance().getAppVersion();
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional string appVersion = 3;</code>
+       * @param value The bytes for appVersion to set.
+       * @return This builder for chaining.
+       */
+      public Builder setAppVersionBytes(
+          akka.protobufv3.internal.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  bitField0_ |= 0x00000004;
+        appVersion_ = value;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final akka.protobufv3.internal.UnknownFieldSet unknownFields) {
@@ -8342,6 +8521,31 @@ public final class ClusterMessages {
      */
     akka.cluster.protobuf.msg.ClusterMessages.TombstoneOrBuilder getTombstonesOrBuilder(
         int index);
+
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @return A list containing the allAppVersions.
+     */
+    java.util.List<java.lang.String>
+        getAllAppVersionsList();
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @return The count of allAppVersions.
+     */
+    int getAllAppVersionsCount();
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @param index The index of the element to return.
+     * @return The allAppVersions at the given index.
+     */
+    java.lang.String getAllAppVersions(int index);
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @param index The index of the value to return.
+     * @return The bytes of the allAppVersions at the given index.
+     */
+    akka.protobufv3.internal.ByteString
+        getAllAppVersionsBytes(int index);
   }
   /**
    * <pre>
@@ -8366,6 +8570,7 @@ public final class ClusterMessages {
       allHashes_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
       members_ = java.util.Collections.emptyList();
       tombstones_ = java.util.Collections.emptyList();
+      allAppVersions_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
     }
 
     @java.lang.Override
@@ -8470,6 +8675,15 @@ public final class ClusterMessages {
                   input.readMessage(akka.cluster.protobuf.msg.ClusterMessages.Tombstone.PARSER, extensionRegistry));
               break;
             }
+            case 66: {
+              akka.protobufv3.internal.ByteString bs = input.readBytes();
+              if (!((mutable_bitField0_ & 0x00000080) != 0)) {
+                allAppVersions_ = new akka.protobufv3.internal.LazyStringArrayList();
+                mutable_bitField0_ |= 0x00000080;
+              }
+              allAppVersions_.add(bs);
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -8500,6 +8714,9 @@ public final class ClusterMessages {
         if (((mutable_bitField0_ & 0x00000040) != 0)) {
           tombstones_ = java.util.Collections.unmodifiableList(tombstones_);
         }
+        if (((mutable_bitField0_ & 0x00000080) != 0)) {
+          allAppVersions_ = allAppVersions_.getUnmodifiableView();
+        }
         this.unknownFields = unknownFields.build();
         makeExtensionsImmutable();
       }
@@ -8739,6 +8956,41 @@ public final class ClusterMessages {
       return tombstones_.get(index);
     }
 
+    public static final int ALLAPPVERSIONS_FIELD_NUMBER = 8;
+    private akka.protobufv3.internal.LazyStringList allAppVersions_;
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @return A list containing the allAppVersions.
+     */
+    public akka.protobufv3.internal.ProtocolStringList
+        getAllAppVersionsList() {
+      return allAppVersions_;
+    }
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @return The count of allAppVersions.
+     */
+    public int getAllAppVersionsCount() {
+      return allAppVersions_.size();
+    }
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @param index The index of the element to return.
+     * @return The allAppVersions at the given index.
+     */
+    public java.lang.String getAllAppVersions(int index) {
+      return allAppVersions_.get(index);
+    }
+    /**
+     * <code>repeated string allAppVersions = 8;</code>
+     * @param index The index of the value to return.
+     * @return The bytes of the allAppVersions at the given index.
+     */
+    public akka.protobufv3.internal.ByteString
+        getAllAppVersionsBytes(int index) {
+      return allAppVersions_.getByteString(index);
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -8808,6 +9060,9 @@ public final class ClusterMessages {
       for (int i = 0; i < tombstones_.size(); i++) {
         output.writeMessage(7, tombstones_.get(i));
       }
+      for (int i = 0; i < allAppVersions_.size(); i++) {
+        akka.protobufv3.internal.GeneratedMessageV3.writeString(output, 8, allAppVersions_.getRaw(i));
+      }
       unknownFields.writeTo(output);
     }
 
@@ -8853,6 +9108,14 @@ public final class ClusterMessages {
         size += akka.protobufv3.internal.CodedOutputStream
           .computeMessageSize(7, tombstones_.get(i));
       }
+      {
+        int dataSize = 0;
+        for (int i = 0; i < allAppVersions_.size(); i++) {
+          dataSize += computeStringSizeNoTag(allAppVersions_.getRaw(i));
+        }
+        size += dataSize;
+        size += 1 * getAllAppVersionsList().size();
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -8888,6 +9151,8 @@ public final class ClusterMessages {
       }
       if (!getTombstonesList()
           .equals(other.getTombstonesList())) return false;
+      if (!getAllAppVersionsList()
+          .equals(other.getAllAppVersionsList())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -8927,6 +9192,10 @@ public final class ClusterMessages {
         hash = (37 * hash) + TOMBSTONES_FIELD_NUMBER;
         hash = (53 * hash) + getTombstonesList().hashCode();
       }
+      if (getAllAppVersionsCount() > 0) {
+        hash = (37 * hash) + ALLAPPVERSIONS_FIELD_NUMBER;
+        hash = (53 * hash) + getAllAppVersionsList().hashCode();
+      }
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -9104,6 +9373,8 @@ public final class ClusterMessages {
         } else {
           tombstonesBuilder_.clear();
         }
+        allAppVersions_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
+        bitField0_ = (bitField0_ & ~0x00000080);
         return this;
       }
 
@@ -9185,6 +9456,11 @@ public final class ClusterMessages {
         } else {
           result.tombstones_ = tombstonesBuilder_.build();
         }
+        if (((bitField0_ & 0x00000080) != 0)) {
+          allAppVersions_ = allAppVersions_.getUnmodifiableView();
+          bitField0_ = (bitField0_ & ~0x00000080);
+        }
+        result.allAppVersions_ = allAppVersions_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -9338,6 +9614,16 @@ public final class ClusterMessages {
             }
           }
         }
+        if (!other.allAppVersions_.isEmpty()) {
+          if (allAppVersions_.isEmpty()) {
+            allAppVersions_ = other.allAppVersions_;
+            bitField0_ = (bitField0_ & ~0x00000080);
+          } else {
+            ensureAllAppVersionsIsMutable();
+            allAppVersions_.addAll(other.allAppVersions_);
+          }
+          onChanged();
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -10572,6 +10858,115 @@ public final class ClusterMessages {
         }
         return tombstonesBuilder_;
       }
+
+      private akka.protobufv3.internal.LazyStringList allAppVersions_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
+      private void ensureAllAppVersionsIsMutable() {
+        if (!((bitField0_ & 0x00000080) != 0)) {
+          allAppVersions_ = new akka.protobufv3.internal.LazyStringArrayList(allAppVersions_);
+          bitField0_ |= 0x00000080;
+         }
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @return A list containing the allAppVersions.
+       */
+      public akka.protobufv3.internal.ProtocolStringList
+          getAllAppVersionsList() {
+        return allAppVersions_.getUnmodifiableView();
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @return The count of allAppVersions.
+       */
+      public int getAllAppVersionsCount() {
+        return allAppVersions_.size();
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @param index The index of the element to return.
+       * @return The allAppVersions at the given index.
+       */
+      public java.lang.String getAllAppVersions(int index) {
+        return allAppVersions_.get(index);
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @param index The index of the value to return.
+       * @return The bytes of the allAppVersions at the given index.
+       */
+      public akka.protobufv3.internal.ByteString
+          getAllAppVersionsBytes(int index) {
+        return allAppVersions_.getByteString(index);
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @param index The index to set the value at.
+       * @param value The allAppVersions to set.
+       * @return This builder for chaining.
+       */
+      public Builder setAllAppVersions(
+          int index, java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  ensureAllAppVersionsIsMutable();
+        allAppVersions_.set(index, value);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @param value The allAppVersions to add.
+       * @return This builder for chaining.
+       */
+      public Builder addAllAppVersions(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  ensureAllAppVersionsIsMutable();
+        allAppVersions_.add(value);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @param values The allAppVersions to add.
+       * @return This builder for chaining.
+       */
+      public Builder addAllAllAppVersions(
+          java.lang.Iterable<java.lang.String> values) {
+        ensureAllAppVersionsIsMutable();
+        akka.protobufv3.internal.AbstractMessageLite.Builder.addAll(
+            values, allAppVersions_);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @return This builder for chaining.
+       */
+      public Builder clearAllAppVersions() {
+        allAppVersions_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
+        bitField0_ = (bitField0_ & ~0x00000080);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string allAppVersions = 8;</code>
+       * @param value The bytes of the allAppVersions to add.
+       * @return This builder for chaining.
+       */
+      public Builder addAllAppVersionsBytes(
+          akka.protobufv3.internal.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  ensureAllAppVersionsIsMutable();
+        allAppVersions_.add(value);
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final akka.protobufv3.internal.UnknownFieldSet unknownFields) {
@@ -14112,6 +14507,17 @@ public final class ClusterMessages {
      * @return The rolesIndexes at the given index.
      */
     int getRolesIndexes(int index);
+
+    /**
+     * <code>optional int32 appVersionIndex = 5;</code>
+     * @return Whether the appVersionIndex field is set.
+     */
+    boolean hasAppVersionIndex();
+    /**
+     * <code>optional int32 appVersionIndex = 5;</code>
+     * @return The appVersionIndex.
+     */
+    int getAppVersionIndex();
   }
   /**
    * <pre>
@@ -14209,6 +14615,11 @@ public final class ClusterMessages {
               input.popLimit(limit);
               break;
             }
+            case 40: {
+              bitField0_ |= 0x00000008;
+              appVersionIndex_ = input.readInt32();
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -14325,6 +14736,23 @@ public final class ClusterMessages {
     }
     private int rolesIndexesMemoizedSerializedSize = -1;
 
+    public static final int APPVERSIONINDEX_FIELD_NUMBER = 5;
+    private int appVersionIndex_;
+    /**
+     * <code>optional int32 appVersionIndex = 5;</code>
+     * @return Whether the appVersionIndex field is set.
+     */
+    public boolean hasAppVersionIndex() {
+      return ((bitField0_ & 0x00000008) != 0);
+    }
+    /**
+     * <code>optional int32 appVersionIndex = 5;</code>
+     * @return The appVersionIndex.
+     */
+    public int getAppVersionIndex() {
+      return appVersionIndex_;
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -14368,6 +14796,9 @@ public final class ClusterMessages {
       for (int i = 0; i < rolesIndexes_.size(); i++) {
         output.writeInt32NoTag(rolesIndexes_.getInt(i));
       }
+      if (((bitField0_ & 0x00000008) != 0)) {
+        output.writeInt32(5, appVersionIndex_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -14403,6 +14834,10 @@ public final class ClusterMessages {
         }
         rolesIndexesMemoizedSerializedSize = dataSize;
       }
+      if (((bitField0_ & 0x00000008) != 0)) {
+        size += akka.protobufv3.internal.CodedOutputStream
+          .computeInt32Size(5, appVersionIndex_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -14434,6 +14869,11 @@ public final class ClusterMessages {
       }
       if (!getRolesIndexesList()
           .equals(other.getRolesIndexesList())) return false;
+      if (hasAppVersionIndex() != other.hasAppVersionIndex()) return false;
+      if (hasAppVersionIndex()) {
+        if (getAppVersionIndex()
+            != other.getAppVersionIndex()) return false;
+      }
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -14461,6 +14901,10 @@ public final class ClusterMessages {
         hash = (37 * hash) + ROLESINDEXES_FIELD_NUMBER;
         hash = (53 * hash) + getRolesIndexesList().hashCode();
       }
+      if (hasAppVersionIndex()) {
+        hash = (37 * hash) + APPVERSIONINDEX_FIELD_NUMBER;
+        hash = (53 * hash) + getAppVersionIndex();
+      }
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -14607,6 +15051,8 @@ public final class ClusterMessages {
         bitField0_ = (bitField0_ & ~0x00000004);
         rolesIndexes_ = emptyIntList();
         bitField0_ = (bitField0_ & ~0x00000008);
+        appVersionIndex_ = 0;
+        bitField0_ = (bitField0_ & ~0x00000010);
         return this;
       }
 
@@ -14652,6 +15098,10 @@ public final class ClusterMessages {
           bitField0_ = (bitField0_ & ~0x00000008);
         }
         result.rolesIndexes_ = rolesIndexes_;
+        if (((from_bitField0_ & 0x00000010) != 0)) {
+          result.appVersionIndex_ = appVersionIndex_;
+          to_bitField0_ |= 0x00000008;
+        }
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -14720,6 +15170,9 @@ public final class ClusterMessages {
           }
           onChanged();
         }
+        if (other.hasAppVersionIndex()) {
+          setAppVersionIndex(other.getAppVersionIndex());
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -14953,6 +15406,43 @@ public final class ClusterMessages {
         onChanged();
         return this;
       }
+
+      private int appVersionIndex_ ;
+      /**
+       * <code>optional int32 appVersionIndex = 5;</code>
+       * @return Whether the appVersionIndex field is set.
+       */
+      public boolean hasAppVersionIndex() {
+        return ((bitField0_ & 0x00000010) != 0);
+      }
+      /**
+       * <code>optional int32 appVersionIndex = 5;</code>
+       * @return The appVersionIndex.
+       */
+      public int getAppVersionIndex() {
+        return appVersionIndex_;
+      }
+      /**
+       * <code>optional int32 appVersionIndex = 5;</code>
+       * @param value The appVersionIndex to set.
+       * @return This builder for chaining.
+       */
+      public Builder setAppVersionIndex(int value) {
+        bitField0_ |= 0x00000010;
+        appVersionIndex_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional int32 appVersionIndex = 5;</code>
+       * @return This builder for chaining.
+       */
+      public Builder clearAppVersionIndex() {
+        bitField0_ = (bitField0_ & ~0x00000010);
+        appVersionIndex_ = 0;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final akka.protobufv3.internal.UnknownFieldSet unknownFields) {
@@ -21902,62 +22392,64 @@ public final class ClusterMessages {
       descriptor;
   static {
     java.lang.String[] descriptorData = {
-      "\n\025ClusterMessages.proto\"3\n\004Join\022\034\n\004node\030" +
-      "\001 \002(\0132\016.UniqueAddress\022\r\n\005roles\030\002 \003(\t\"@\n\007" +
-      "Welcome\022\034\n\004from\030\001 \002(\0132\016.UniqueAddress\022\027\n" +
-      "\006gossip\030\002 \002(\0132\007.Gossip\"!\n\010InitJoin\022\025\n\rcu" +
-      "rrentConfig\030\001 \001(\t\"K\n\013InitJoinAck\022\031\n\007addr" +
-      "ess\030\001 \002(\0132\010.Address\022!\n\013configCheck\030\002 \002(\013" +
-      "2\014.ConfigCheck\"\220\001\n\013ConfigCheck\022\037\n\004type\030\001" +
-      " \002(\0162\021.ConfigCheck.Type\022\025\n\rclusterConfig" +
-      "\030\002 \001(\t\"I\n\004Type\022\023\n\017UncheckedConfig\020\001\022\026\n\022I" +
-      "ncompatibleConfig\020\002\022\024\n\020CompatibleConfig\020" +
-      "\003\"M\n\tHeartbeat\022\026\n\004from\030\001 \002(\0132\010.Address\022\022" +
-      "\n\nsequenceNr\030\002 \001(\003\022\024\n\014creationTime\030\003 \001(\022" +
-      "\"[\n\021HeartBeatResponse\022\034\n\004from\030\001 \002(\0132\016.Un" +
-      "iqueAddress\022\022\n\nsequenceNr\030\002 \001(\003\022\024\n\014creat" +
-      "ionTime\030\003 \001(\003\"d\n\016GossipEnvelope\022\034\n\004from\030" +
-      "\001 \002(\0132\016.UniqueAddress\022\032\n\002to\030\002 \002(\0132\016.Uniq" +
-      "ueAddress\022\030\n\020serializedGossip\030\003 \002(\014\"r\n\014G" +
-      "ossipStatus\022\034\n\004from\030\001 \002(\0132\016.UniqueAddres" +
-      "s\022\021\n\tallHashes\030\002 \003(\t\022\035\n\007version\030\003 \002(\0132\014." +
-      "VectorClock\022\022\n\nseenDigest\030\004 \001(\014\"\317\001\n\006Goss" +
-      "ip\022$\n\014allAddresses\030\001 \003(\0132\016.UniqueAddress" +
-      "\022\020\n\010allRoles\030\002 \003(\t\022\021\n\tallHashes\030\003 \003(\t\022\030\n" +
-      "\007members\030\004 \003(\0132\007.Member\022!\n\010overview\030\005 \002(" +
-      "\0132\017.GossipOverview\022\035\n\007version\030\006 \002(\0132\014.Ve" +
-      "ctorClock\022\036\n\ntombstones\030\007 \003(\0132\n.Tombston" +
-      "e\"S\n\016GossipOverview\022\014\n\004seen\030\001 \003(\005\0223\n\024obs" +
-      "erverReachability\030\002 \003(\0132\025.ObserverReacha" +
-      "bility\"p\n\024ObserverReachability\022\024\n\014addres" +
-      "sIndex\030\001 \002(\005\022\017\n\007version\030\004 \002(\003\0221\n\023subject" +
-      "Reachability\030\002 \003(\0132\024.SubjectReachability" +
-      "\"a\n\023SubjectReachability\022\024\n\014addressIndex\030" +
-      "\001 \002(\005\022#\n\006status\030\003 \002(\0162\023.ReachabilityStat" +
-      "us\022\017\n\007version\030\004 \002(\003\"4\n\tTombstone\022\024\n\014addr" +
-      "essIndex\030\001 \002(\005\022\021\n\ttimestamp\030\002 \002(\003\"i\n\006Mem" +
-      "ber\022\024\n\014addressIndex\030\001 \002(\005\022\020\n\010upNumber\030\002 " +
-      "\002(\005\022\035\n\006status\030\003 \002(\0162\r.MemberStatus\022\030\n\014ro" +
-      "lesIndexes\030\004 \003(\005B\002\020\001\"y\n\013VectorClock\022\021\n\tt" +
-      "imestamp\030\001 \001(\003\022&\n\010versions\030\002 \003(\0132\024.Vecto" +
-      "rClock.Version\032/\n\007Version\022\021\n\thashIndex\030\001" +
-      " \002(\005\022\021\n\ttimestamp\030\002 \002(\003\"\007\n\005Empty\"K\n\007Addr" +
-      "ess\022\016\n\006system\030\001 \002(\t\022\020\n\010hostname\030\002 \002(\t\022\014\n" +
-      "\004port\030\003 \002(\r\022\020\n\010protocol\030\004 \001(\t\"E\n\rUniqueA" +
-      "ddress\022\031\n\007address\030\001 \002(\0132\010.Address\022\013\n\003uid" +
-      "\030\002 \002(\r\022\014\n\004uid2\030\003 \001(\r\"V\n\021ClusterRouterPoo" +
-      "l\022\023\n\004pool\030\001 \002(\0132\005.Pool\022,\n\010settings\030\002 \002(\013" +
-      "2\032.ClusterRouterPoolSettings\"<\n\004Pool\022\024\n\014" +
-      "serializerId\030\001 \002(\r\022\020\n\010manifest\030\002 \002(\t\022\014\n\004" +
-      "data\030\003 \002(\014\"\216\001\n\031ClusterRouterPoolSettings" +
-      "\022\026\n\016totalInstances\030\001 \002(\r\022\033\n\023maxInstances" +
-      "PerNode\030\002 \002(\r\022\031\n\021allowLocalRoutees\030\003 \002(\010" +
-      "\022\017\n\007useRole\030\004 \001(\t\022\020\n\010useRoles\030\005 \003(\t*D\n\022R" +
-      "eachabilityStatus\022\r\n\tReachable\020\000\022\017\n\013Unre" +
-      "achable\020\001\022\016\n\nTerminated\020\002*b\n\014MemberStatu" +
-      "s\022\013\n\007Joining\020\000\022\006\n\002Up\020\001\022\013\n\007Leaving\020\002\022\013\n\007E" +
-      "xiting\020\003\022\010\n\004Down\020\004\022\013\n\007Removed\020\005\022\014\n\010Weakl" +
-      "yUp\020\006B\035\n\031akka.cluster.protobuf.msgH\001"
+      "\n\025ClusterMessages.proto\"G\n\004Join\022\034\n\004node\030" +
+      "\001 \002(\0132\016.UniqueAddress\022\r\n\005roles\030\002 \003(\t\022\022\n\n" +
+      "appVersion\030\003 \001(\t\"@\n\007Welcome\022\034\n\004from\030\001 \002(" +
+      "\0132\016.UniqueAddress\022\027\n\006gossip\030\002 \002(\0132\007.Goss" +
+      "ip\"!\n\010InitJoin\022\025\n\rcurrentConfig\030\001 \001(\t\"K\n" +
+      "\013InitJoinAck\022\031\n\007address\030\001 \002(\0132\010.Address\022" +
+      "!\n\013configCheck\030\002 \002(\0132\014.ConfigCheck\"\220\001\n\013C" +
+      "onfigCheck\022\037\n\004type\030\001 \002(\0162\021.ConfigCheck.T" +
+      "ype\022\025\n\rclusterConfig\030\002 \001(\t\"I\n\004Type\022\023\n\017Un" +
+      "checkedConfig\020\001\022\026\n\022IncompatibleConfig\020\002\022" +
+      "\024\n\020CompatibleConfig\020\003\"M\n\tHeartbeat\022\026\n\004fr" +
+      "om\030\001 \002(\0132\010.Address\022\022\n\nsequenceNr\030\002 \001(\003\022\024" +
+      "\n\014creationTime\030\003 \001(\022\"[\n\021HeartBeatRespons" +
+      "e\022\034\n\004from\030\001 \002(\0132\016.UniqueAddress\022\022\n\nseque" +
+      "nceNr\030\002 \001(\003\022\024\n\014creationTime\030\003 \001(\003\"d\n\016Gos" +
+      "sipEnvelope\022\034\n\004from\030\001 \002(\0132\016.UniqueAddres" +
+      "s\022\032\n\002to\030\002 \002(\0132\016.UniqueAddress\022\030\n\020seriali" +
+      "zedGossip\030\003 \002(\014\"r\n\014GossipStatus\022\034\n\004from\030" +
+      "\001 \002(\0132\016.UniqueAddress\022\021\n\tallHashes\030\002 \003(\t" +
+      "\022\035\n\007version\030\003 \002(\0132\014.VectorClock\022\022\n\nseenD" +
+      "igest\030\004 \001(\014\"\347\001\n\006Gossip\022$\n\014allAddresses\030\001" +
+      " \003(\0132\016.UniqueAddress\022\020\n\010allRoles\030\002 \003(\t\022\021" +
+      "\n\tallHashes\030\003 \003(\t\022\030\n\007members\030\004 \003(\0132\007.Mem" +
+      "ber\022!\n\010overview\030\005 \002(\0132\017.GossipOverview\022\035" +
+      "\n\007version\030\006 \002(\0132\014.VectorClock\022\036\n\ntombsto" +
+      "nes\030\007 \003(\0132\n.Tombstone\022\026\n\016allAppVersions\030" +
+      "\010 \003(\t\"S\n\016GossipOverview\022\014\n\004seen\030\001 \003(\005\0223\n" +
+      "\024observerReachability\030\002 \003(\0132\025.ObserverRe" +
+      "achability\"p\n\024ObserverReachability\022\024\n\014ad" +
+      "dressIndex\030\001 \002(\005\022\017\n\007version\030\004 \002(\003\0221\n\023sub" +
+      "jectReachability\030\002 \003(\0132\024.SubjectReachabi" +
+      "lity\"a\n\023SubjectReachability\022\024\n\014addressIn" +
+      "dex\030\001 \002(\005\022#\n\006status\030\003 \002(\0162\023.Reachability" +
+      "Status\022\017\n\007version\030\004 \002(\003\"4\n\tTombstone\022\024\n\014" +
+      "addressIndex\030\001 \002(\005\022\021\n\ttimestamp\030\002 \002(\003\"\202\001" +
+      "\n\006Member\022\024\n\014addressIndex\030\001 \002(\005\022\020\n\010upNumb" +
+      "er\030\002 \002(\005\022\035\n\006status\030\003 \002(\0162\r.MemberStatus\022" +
+      "\030\n\014rolesIndexes\030\004 \003(\005B\002\020\001\022\027\n\017appVersionI" +
+      "ndex\030\005 \001(\005\"y\n\013VectorClock\022\021\n\ttimestamp\030\001" +
+      " \001(\003\022&\n\010versions\030\002 \003(\0132\024.VectorClock.Ver" +
+      "sion\032/\n\007Version\022\021\n\thashIndex\030\001 \002(\005\022\021\n\tti" +
+      "mestamp\030\002 \002(\003\"\007\n\005Empty\"K\n\007Address\022\016\n\006sys" +
+      "tem\030\001 \002(\t\022\020\n\010hostname\030\002 \002(\t\022\014\n\004port\030\003 \002(" +
+      "\r\022\020\n\010protocol\030\004 \001(\t\"E\n\rUniqueAddress\022\031\n\007" +
+      "address\030\001 \002(\0132\010.Address\022\013\n\003uid\030\002 \002(\r\022\014\n\004" +
+      "uid2\030\003 \001(\r\"V\n\021ClusterRouterPool\022\023\n\004pool\030" +
+      "\001 \002(\0132\005.Pool\022,\n\010settings\030\002 \002(\0132\032.Cluster" +
+      "RouterPoolSettings\"<\n\004Pool\022\024\n\014serializer" +
+      "Id\030\001 \002(\r\022\020\n\010manifest\030\002 \002(\t\022\014\n\004data\030\003 \002(\014" +
+      "\"\216\001\n\031ClusterRouterPoolSettings\022\026\n\016totalI" +
+      "nstances\030\001 \002(\r\022\033\n\023maxInstancesPerNode\030\002 " +
+      "\002(\r\022\031\n\021allowLocalRoutees\030\003 \002(\010\022\017\n\007useRol" +
+      "e\030\004 \001(\t\022\020\n\010useRoles\030\005 \003(\t*D\n\022Reachabilit" +
+      "yStatus\022\r\n\tReachable\020\000\022\017\n\013Unreachable\020\001\022" +
+      "\016\n\nTerminated\020\002*b\n\014MemberStatus\022\013\n\007Joini" +
+      "ng\020\000\022\006\n\002Up\020\001\022\013\n\007Leaving\020\002\022\013\n\007Exiting\020\003\022\010" +
+      "\n\004Down\020\004\022\013\n\007Removed\020\005\022\014\n\010WeaklyUp\020\006B\035\n\031a" +
+      "kka.cluster.protobuf.msgH\001"
     };
     descriptor = akka.protobufv3.internal.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -21968,7 +22460,7 @@ public final class ClusterMessages {
     internal_static_Join_fieldAccessorTable = new
       akka.protobufv3.internal.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Join_descriptor,
-        new java.lang.String[] { "Node", "Roles", });
+        new java.lang.String[] { "Node", "Roles", "AppVersion", });
     internal_static_Welcome_descriptor =
       getDescriptor().getMessageTypes().get(1);
     internal_static_Welcome_fieldAccessorTable = new
@@ -22022,7 +22514,7 @@ public final class ClusterMessages {
     internal_static_Gossip_fieldAccessorTable = new
       akka.protobufv3.internal.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Gossip_descriptor,
-        new java.lang.String[] { "AllAddresses", "AllRoles", "AllHashes", "Members", "Overview", "Version", "Tombstones", });
+        new java.lang.String[] { "AllAddresses", "AllRoles", "AllHashes", "Members", "Overview", "Version", "Tombstones", "AllAppVersions", });
     internal_static_GossipOverview_descriptor =
       getDescriptor().getMessageTypes().get(10);
     internal_static_GossipOverview_fieldAccessorTable = new
@@ -22052,7 +22544,7 @@ public final class ClusterMessages {
     internal_static_Member_fieldAccessorTable = new
       akka.protobufv3.internal.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Member_descriptor,
-        new java.lang.String[] { "AddressIndex", "UpNumber", "Status", "RolesIndexes", });
+        new java.lang.String[] { "AddressIndex", "UpNumber", "Status", "RolesIndexes", "AppVersionIndex", });
     internal_static_VectorClock_descriptor =
       getDescriptor().getMessageTypes().get(15);
     internal_static_VectorClock_fieldAccessorTable = new
diff --git a/akka-cluster/src/main/mima-filters/2.6.8.backwards.excludes/issue-27300-appVersion.excludes b/akka-cluster/src/main/mima-filters/2.6.8.backwards.excludes/issue-27300-appVersion.excludes
new file mode 100644
index 0000000000..57e176ca29
--- /dev/null
+++ b/akka-cluster/src/main/mima-filters/2.6.8.backwards.excludes/issue-27300-appVersion.excludes
@@ -0,0 +1,9 @@
+# #27300 Added appVersion to Member
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.InternalClusterAction#Join.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.InternalClusterAction#Join.this")
+ProblemFilters.exclude[MissingTypesProblem]("akka.cluster.InternalClusterAction$Join$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.InternalClusterAction#Join.apply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.cluster.InternalClusterAction#Join.unapply")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Member.apply")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Member.this")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterCoreDaemon.joining")
diff --git a/akka-cluster/src/main/protobuf/ClusterMessages.proto b/akka-cluster/src/main/protobuf/ClusterMessages.proto
index 04a77f2e6a..fe7885538a 100644
--- a/akka-cluster/src/main/protobuf/ClusterMessages.proto
+++ b/akka-cluster/src/main/protobuf/ClusterMessages.proto
@@ -31,6 +31,7 @@ option optimize_for = SPEED;
 message Join {
   required UniqueAddress node = 1;
   repeated string roles = 2;
+  optional string appVersion = 3;
 }
 
 /**
@@ -133,6 +134,7 @@ message Gossip {
   required GossipOverview overview = 5;
   required VectorClock version = 6;
   repeated Tombstone tombstones = 7;
+  repeated string allAppVersions = 8;
 }
 
 /**
@@ -182,6 +184,7 @@ message Member {
   required int32 upNumber = 2;
   required MemberStatus status = 3;
   repeated int32 rolesIndexes = 4 [packed = true];
+  optional int32 appVersionIndex = 5;
 }
 
 /**
diff --git a/akka-cluster/src/main/resources/reference.conf b/akka-cluster/src/main/resources/reference.conf
index 53b29c2210..e067241524 100644
--- a/akka-cluster/src/main/resources/reference.conf
+++ b/akka-cluster/src/main/resources/reference.conf
@@ -93,6 +93,23 @@ akka {
       #<role-name>.min-nr-of-members = 1
     }
 
+    # Application version of the deployment. Used by rolling update features
+    # to distinguish between old and new nodes. The typical convention is to use
+    # 3 digit version numbers `major.minor.patch`, but 1 or two digits are also
+    # supported.
+    #
+    # If no `.` is used it is interpreted as a single digit version number or as
+    # plain alphanumeric if it couldn't be parsed as a number.
+    #
+    # It may also have a qualifier at the end for 2 or 3 digit version numbers such
+    # as "1.2-RC1".
+    # For 1 digit with qualifier, 1-RC1, it is interpreted as plain alphanumeric.
+    #
+    # It has support for https://github.com/dwijnand/sbt-dynver format with `+` or
+    # `-` separator. The number of commits from the tag is handled as a numeric part.
+    # For example `1.0.0+3-73475dce26` is less than `1.0.10+10-ed316bd024` (3 < 10).
+    app-version = "0.0.0"
+
     # Minimum required number of members before the leader changes member status
     # of 'Joining' members to 'Up'. Typically used together with
     # 'Cluster.registerOnMemberUp' to defer some action, such as starting actors,
diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala
index 03e52848a6..0d535bd2cb 100644
--- a/akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala
@@ -26,6 +26,7 @@ import akka.pattern.ask
 import akka.remote.{ QuarantinedEvent => ClassicQuarantinedEvent }
 import akka.remote.artery.QuarantinedEvent
 import akka.util.Timeout
+import akka.util.Version
 
 /**
  * Base trait for all cluster messages. All ClusterMessage's are serializable.
@@ -74,7 +75,7 @@ private[cluster] object InternalClusterAction {
    * @param node the node that wants to join the cluster
    */
   @SerialVersionUID(1L)
-  final case class Join(node: UniqueAddress, roles: Set[String]) extends ClusterMessage
+  final case class Join(node: UniqueAddress, roles: Set[String], appVersion: Version) extends ClusterMessage
 
   /**
    * Reply to Join
@@ -557,7 +558,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
       case InitJoin(joiningNodeConfig) =>
         logInfo("Received InitJoin message from [{}] to [{}]", sender(), selfAddress)
         initJoin(joiningNodeConfig)
-      case Join(node, roles)                     => joining(node, roles)
+      case Join(node, roles, appVersion)         => joining(node, roles, appVersion)
       case ClusterUserAction.Down(address)       => downing(address)
       case ClusterUserAction.Leave(address)      => leaving(address)
       case SendGossipTo(address)                 => sendGossipTo(address)
@@ -705,14 +706,14 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
 
       if (address == selfAddress) {
         becomeInitialized()
-        joining(selfUniqueAddress, cluster.selfRoles)
+        joining(selfUniqueAddress, cluster.selfRoles, cluster.settings.AppVersion)
       } else {
         val joinDeadline = RetryUnsuccessfulJoinAfter match {
           case d: FiniteDuration => Some(Deadline.now + d)
           case _                 => None
         }
         context.become(tryingToJoin(address, joinDeadline))
-        clusterCore(address) ! Join(selfUniqueAddress, cluster.selfRoles)
+        clusterCore(address) ! Join(selfUniqueAddress, cluster.selfRoles, cluster.settings.AppVersion)
       }
     }
   }
@@ -732,7 +733,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
    * Received `Join` message and replies with `Welcome` message, containing
    * current gossip state, including the new joining member.
    */
-  def joining(joiningNode: UniqueAddress, roles: Set[String]): Unit = {
+  def joining(joiningNode: UniqueAddress, roles: Set[String], appVersion: Version): Unit = {
     val selfStatus = latestGossip.member(selfUniqueAddress).status
     if (joiningNode.address.protocol != selfAddress.protocol)
       logWarning(
@@ -781,7 +782,10 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
 
           // add joining node as Joining
           // add self in case someone else joins before self has joined (Set discards duplicates)
-          val newMembers = localMembers + Member(joiningNode, roles) + Member(selfUniqueAddress, cluster.selfRoles)
+          val newMembers = localMembers + Member(joiningNode, roles, appVersion) + Member(
+              selfUniqueAddress,
+              cluster.selfRoles,
+              cluster.settings.AppVersion)
           val newGossip = latestGossip.copy(members = newMembers)
 
           updateLatestGossip(newGossip)
@@ -789,17 +793,19 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
           if (joiningNode == selfUniqueAddress) {
             logInfo(
               ClusterLogMarker.memberChanged(joiningNode, MemberStatus.Joining),
-              "Node [{}] is JOINING itself (with roles [{}]) and forming new cluster",
+              "Node [{}] is JOINING itself (with roles [{}], version [{}]) and forming new cluster",
               joiningNode.address,
-              roles.mkString(", "))
+              roles.mkString(", "),
+              appVersion)
             if (localMembers.isEmpty)
               leaderActions() // important for deterministic oldest when bootstrapping
           } else {
             logInfo(
               ClusterLogMarker.memberChanged(joiningNode, MemberStatus.Joining),
-              "Node [{}] is JOINING, roles [{}]",
+              "Node [{}] is JOINING, roles [{}], version [{}]",
               joiningNode.address,
-              roles.mkString(", "))
+              roles.mkString(", "),
+              appVersion)
             sender() ! Welcome(selfUniqueAddress, latestGossip)
           }
 
diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala
index a4786c7cbb..830cfbb15e 100644
--- a/akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala
@@ -187,6 +187,19 @@ object ClusterEvent {
     def getAllDataCenters: java.util.Set[String] =
       scala.collection.JavaConverters.setAsJavaSetConverter(allDataCenters).asJava
 
+    /**
+     * INTERNAL API
+     * @return `true` if more than one `Version` among the members, which
+     *        indicates that a rolling update is in progress
+     */
+    @InternalApi private[akka] def hasMoreThanOneAppVersion: Boolean = {
+      if (members.isEmpty) false
+      else {
+        val v = members.head.appVersion
+        members.exists(_.appVersion != v)
+      }
+    }
+
     /**
      * Replace the set of unreachable datacenters with the given set
      */
diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterJmx.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterJmx.scala
index 1d3ad80003..c9eaeb7f44 100644
--- a/akka-cluster/src/main/scala/akka/cluster/ClusterJmx.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/ClusterJmx.scala
@@ -44,6 +44,7 @@ trait ClusterNodeMBean {
    *     {
    *       "address": "akka://system@host1:2552",
    *       "status": "Up",
+   *       "app-version": "1.0.0",
    *       "roles": [
    *         "frontend"
    *       ]
@@ -51,6 +52,7 @@ trait ClusterNodeMBean {
    *     {
    *       "address": "akka://system@host2:2552",
    *       "status": "Up",
+   *       "app-version": "1.0.0",
    *       "roles": [
    *         "frontend"
    *       ]
@@ -58,6 +60,7 @@ trait ClusterNodeMBean {
    *     {
    *       "address": "akka://system@host3:2552",
    *       "status": "Down",
+   *       "app-version": "1.0.0",
    *       "roles": [
    *         "backend"
    *       ]
@@ -65,6 +68,7 @@ trait ClusterNodeMBean {
    *     {
    *       "address": "akka://system@host4:2552",
    *       "status": "Joining",
+   *       "app-version": "1.1.0",
    *       "roles": [
    *         "backend"
    *       ]
@@ -159,7 +163,8 @@ private[akka] class ClusterJmx(cluster: Cluster, log: LoggingAdapter) {
               |      "address": "${m.address}",
               |      "roles": [${if (m.roles.isEmpty) ""
                else m.roles.toList.sorted.map("\"" + _ + "\"").mkString("\n        ", ",\n        ", "\n      ")}],
-              |      "status": "${m.status}"
+              |      "status": "${m.status}",
+              |      "app-version": "${m.appVersion}"
               |    }""".stripMargin
           }
           .mkString(",\n    ")
diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterReadView.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterReadView.scala
index abcdae4004..0ce638a24e 100644
--- a/akka-cluster/src/main/scala/akka/cluster/ClusterReadView.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/ClusterReadView.scala
@@ -38,7 +38,8 @@ private[akka] class ClusterReadView(cluster: Cluster) extends Closeable {
 
   @volatile
   private var _cachedSelf: Member =
-    Member(cluster.selfUniqueAddress, cluster.selfRoles).copy(status = MemberStatus.Removed)
+    Member(cluster.selfUniqueAddress, cluster.selfRoles, cluster.settings.AppVersion)
+      .copy(status = MemberStatus.Removed)
   @volatile
   private var _closed: Boolean = false
 
diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterSettings.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterSettings.scala
index 94bfdc5779..32812e68d2 100644
--- a/akka-cluster/src/main/scala/akka/cluster/ClusterSettings.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/ClusterSettings.scala
@@ -16,6 +16,7 @@ import akka.actor.AddressFromURIString
 import akka.annotation.InternalApi
 import akka.japi.Util.immutableSeq
 import akka.util.Helpers.{ toRootLowerCase, ConfigOps, Requiring }
+import akka.util.Version
 
 object ClusterSettings {
   type DataCenter = String
@@ -146,6 +147,9 @@ final class ClusterSettings(val config: Config, val systemName: String) {
     configuredRoles + s"$DcRolePrefix$SelfDataCenter"
   }
 
+  val AppVersion: Version =
+    Version(cc.getString("app-version"))
+
   val MinNrOfMembers: Int = {
     cc.getInt("min-nr-of-members")
   }.requiring(_ > 0, "min-nr-of-members must be > 0")
diff --git a/akka-cluster/src/main/scala/akka/cluster/Member.scala b/akka-cluster/src/main/scala/akka/cluster/Member.scala
index 1f01f53cbe..d165701a7f 100644
--- a/akka-cluster/src/main/scala/akka/cluster/Member.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/Member.scala
@@ -6,12 +6,13 @@ package akka.cluster
 
 import scala.runtime.AbstractFunction2
 
-import MemberStatus._
 import com.github.ghik.silencer.silent
 
 import akka.actor.Address
 import akka.annotation.InternalApi
 import akka.cluster.ClusterSettings.DataCenter
+import akka.cluster.MemberStatus._
+import akka.util.Version
 
 /**
  * Represents the address, current status, and roles of a cluster member node.
@@ -24,7 +25,8 @@ class Member private[cluster] (
     val uniqueAddress: UniqueAddress,
     private[cluster] val upNumber: Int, // INTERNAL API
     val status: MemberStatus,
-    val roles: Set[String])
+    val roles: Set[String],
+    private[akka] val appVersion: Version) // INTERNAL API
     extends Serializable {
 
   lazy val dataCenter: DataCenter = roles
@@ -34,16 +36,15 @@ class Member private[cluster] (
 
   def address: Address = uniqueAddress.address
 
-  override def hashCode = uniqueAddress.##
-  override def equals(other: Any) = other match {
+  override def hashCode: Int = uniqueAddress.##
+  override def equals(other: Any): Boolean = other match {
     case m: Member => uniqueAddress == m.uniqueAddress
     case _         => false
   }
-  override def toString =
-    if (dataCenter == ClusterSettings.DefaultDataCenter)
-      s"Member(address = $address, status = $status, roles = $roles)"
-    else
-      s"Member(address = $address, dataCenter = $dataCenter, status = $status, roles = $roles)"
+  override def toString: String = {
+    s"Member($address, $status${if (dataCenter == ClusterSettings.DefaultDataCenter) "" else s", $dataCenter"}${if (appVersion == Version.Zero) ""
+    else s", $appVersion"})"
+  }
 
   def hasRole(role: String): Boolean = roles.contains(role)
 
@@ -83,12 +84,12 @@ class Member private[cluster] (
     if (status == oldStatus) this
     else {
       require(allowedTransitions(oldStatus)(status), s"Invalid member status transition [ ${this} -> ${status}]")
-      new Member(uniqueAddress, upNumber, status, roles)
+      new Member(uniqueAddress, upNumber, status, roles, appVersion)
     }
   }
 
   def copyUp(upNumber: Int): Member = {
-    new Member(uniqueAddress, upNumber, status, roles).copy(Up)
+    new Member(uniqueAddress, upNumber, status, roles, appVersion).copy(Up)
   }
 }
 
@@ -103,14 +104,15 @@ object Member {
    * INTERNAL API
    * Create a new member with status Joining.
    */
-  private[akka] def apply(uniqueAddress: UniqueAddress, roles: Set[String]): Member =
-    new Member(uniqueAddress, Int.MaxValue, Joining, roles)
+  @InternalApi
+  private[akka] def apply(uniqueAddress: UniqueAddress, roles: Set[String], appVersion: Version): Member =
+    new Member(uniqueAddress, Int.MaxValue, Joining, roles, appVersion)
 
   /**
    * INTERNAL API
    */
   private[cluster] def removed(node: UniqueAddress): Member =
-    new Member(node, Int.MaxValue, Removed, Set(ClusterSettings.DcRolePrefix + "-N/A"))
+    new Member(node, Int.MaxValue, Removed, Set(ClusterSettings.DcRolePrefix + "-N/A"), Version.Zero)
 
   /**
    * `Address` ordering type class, sorts addresses by host and port.
diff --git a/akka-cluster/src/main/scala/akka/cluster/protobuf/ClusterMessageSerializer.scala b/akka-cluster/src/main/scala/akka/cluster/protobuf/ClusterMessageSerializer.scala
index 6714748d3e..c38f246241 100644
--- a/akka-cluster/src/main/scala/akka/cluster/protobuf/ClusterMessageSerializer.scala
+++ b/akka-cluster/src/main/scala/akka/cluster/protobuf/ClusterMessageSerializer.scala
@@ -23,6 +23,7 @@ import akka.cluster.routing.{ ClusterRouterPool, ClusterRouterPoolSettings }
 import akka.protobufv3.internal.{ ByteString, MessageLite }
 import akka.routing.Pool
 import akka.serialization._
+import akka.util.Version
 import akka.util.ccompat._
 import akka.util.ccompat.JavaConverters._
 
@@ -101,7 +102,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
     case hbr: ClusterHeartbeatSender.HeartbeatRsp                => heartbeatRspToProtoByteArray(hbr)
     case m: GossipEnvelope                                       => gossipEnvelopeToProto(m).toByteArray
     case m: GossipStatus                                         => gossipStatusToProto(m).toByteArray
-    case InternalClusterAction.Join(node, roles)                 => joinToProto(node, roles).toByteArray
+    case InternalClusterAction.Join(node, roles, appVersion)     => joinToProto(node, roles, appVersion).toByteArray
     case InternalClusterAction.Welcome(from, gossip)             => compress(welcomeToProto(from, gossip))
     case ClusterUserAction.Leave(address)                        => addressToProtoByteArray(address)
     case ClusterUserAction.Down(address)                         => addressToProtoByteArray(address)
@@ -275,10 +276,13 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
   private def deserializeJoin(bytes: Array[Byte]): InternalClusterAction.Join = {
     val m = cm.Join.parseFrom(bytes)
     val roles = Set.empty[String] ++ m.getRolesList.asScala
+    // important to use new Version here for lazy parsing
+    val appVersion = if (m.hasAppVersion) new Version(m.getAppVersion) else Version.Zero
     InternalClusterAction.Join(
       uniqueAddressFromProto(m.getNode),
       if (roles.exists(_.startsWith(ClusterSettings.DcRolePrefix))) roles
-      else roles + (ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter))
+      else roles + (ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter),
+      appVersion)
   }
 
   private def deserializeWelcome(bytes: Array[Byte]): InternalClusterAction.Welcome = {
@@ -384,8 +388,13 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
     case _       => throw new IllegalArgumentException(s"Unknown $unknown [$value] in cluster message")
   }
 
-  private def joinToProto(node: UniqueAddress, roles: Set[String]): cm.Join =
-    cm.Join.newBuilder().setNode(uniqueAddressToProto(node)).addAllRoles(roles.asJava).build()
+  private def joinToProto(node: UniqueAddress, roles: Set[String], appVersion: Version): cm.Join =
+    cm.Join
+      .newBuilder()
+      .setNode(uniqueAddressToProto(node))
+      .addAllRoles(roles.asJava)
+      .setAppVersion(appVersion.version)
+      .build()
 
   private def initJoinToProto(currentConfig: Config): cm.InitJoin = {
     cm.InitJoin.newBuilder().setCurrentConfig(currentConfig.root.render(ConfigRenderOptions.concise)).build()
@@ -432,18 +441,24 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
     val roleMapping = allRoles.zipWithIndex.toMap
     val allHashes = gossip.version.versions.keys.to(Vector)
     val hashMapping = allHashes.zipWithIndex.toMap
+    val allAppVersions = allMembers.map(_.appVersion.version)
+    val appVersionMapping = allAppVersions.zipWithIndex.toMap
 
     def mapUniqueAddress(uniqueAddress: UniqueAddress): Integer =
       mapWithErrorMessage(addressMapping, uniqueAddress, "address")
 
     def mapRole(role: String): Integer = mapWithErrorMessage(roleMapping, role, "role")
 
+    def mapAppVersion(appVersion: Version): Integer =
+      mapWithErrorMessage(appVersionMapping, appVersion.version, "appVersion")
+
     def memberToProto(member: Member) =
       cm.Member.newBuilder
         .setAddressIndex(mapUniqueAddress(member.uniqueAddress))
         .setUpNumber(member.upNumber)
         .setStatus(cm.MemberStatus.forNumber(memberStatusToInt(member.status)))
         .addAllRolesIndexes(member.roles.map(mapRole).asJava)
+        .setAppVersionIndex(mapAppVersion(member.appVersion))
 
     def reachabilityToProto(reachability: Reachability): Iterable[cm.ObserverReachability.Builder] = {
       reachability.versions.map {
@@ -484,6 +499,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
       .setOverview(overview)
       .setVersion(vectorClockToProto(gossip.version, hashMapping))
       .addAllTombstones(gossip.tombstones.map(tombstoneToProto _).asJava)
+      .addAllAllAppVersions(allAppVersions.asJava)
   }
 
   private def vectorClockToProto(version: VectorClock, hashMapping: Map[String, Int]): cm.VectorClock.Builder = {
@@ -522,9 +538,11 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
 
   private def gossipFromProto(gossip: cm.Gossip): Gossip = {
     val addressMapping: Vector[UniqueAddress] =
-      gossip.getAllAddressesList.asScala.iterator.map(uniqueAddressFromProto).to(immutable.Vector)
-    val roleMapping: Vector[String] = gossip.getAllRolesList.asScala.iterator.map(identity).to(immutable.Vector)
-    val hashMapping: Vector[String] = gossip.getAllHashesList.asScala.iterator.map(identity).to(immutable.Vector)
+      gossip.getAllAddressesList.asScala.iterator.map(uniqueAddressFromProto).toVector
+    val roleMapping: Vector[String] = gossip.getAllRolesList.asScala.iterator.map(identity).toVector
+    val hashMapping: Vector[String] = gossip.getAllHashesList.asScala.iterator.map(identity).toVector
+    val appVersionMapping: Vector[Version] =
+      gossip.getAllAppVersionsList.asScala.iterator.map(Version(_)).toVector
 
     def reachabilityFromProto(observerReachability: Iterable[cm.ObserverReachability]): Reachability = {
       val recordBuilder = new immutable.VectorBuilder[Reachability.Record]
@@ -548,7 +566,8 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
         addressMapping(member.getAddressIndex),
         member.getUpNumber,
         memberStatusFromInt(member.getStatus.getNumber),
-        rolesFromProto(member.getRolesIndexesList.asScala.toSeq))
+        rolesFromProto(member.getRolesIndexesList.asScala.toSeq),
+        appVersionMapping(member.getAppVersionIndex))
 
     def rolesFromProto(roleIndexes: Seq[Integer]): Set[String] = {
       var containsDc = false
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinSeedNodeSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinSeedNodeSpec.scala
index 9eae814c00..50d5b18eb5 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinSeedNodeSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinSeedNodeSpec.scala
@@ -6,10 +6,13 @@ package akka.cluster
 
 import scala.collection.immutable
 
+import com.typesafe.config.ConfigFactory
+
 import akka.actor.Address
 import akka.remote.testkit.MultiNodeConfig
 import akka.remote.testkit.MultiNodeSpec
 import akka.testkit._
+import akka.util.Version
 
 object JoinSeedNodeMultiJvmSpec extends MultiNodeConfig {
   val seed1 = role("seed1")
@@ -18,7 +21,12 @@ object JoinSeedNodeMultiJvmSpec extends MultiNodeConfig {
   val ordinary1 = role("ordinary1")
   val ordinary2 = role("ordinary2")
 
-  commonConfig(debugConfig(on = false).withFallback(MultiNodeClusterSpec.clusterConfig))
+  commonConfig(
+    debugConfig(on = false)
+      .withFallback(ConfigFactory.parseString("""akka.cluster.app-version="1.0""""))
+      .withFallback(MultiNodeClusterSpec.clusterConfig))
+
+  nodeConfig(ordinary1, ordinary2)(ConfigFactory.parseString("""akka.cluster.app-version="2.0""""))
 }
 
 class JoinSeedNodeMultiJvmNode1 extends JoinSeedNodeSpec
@@ -57,6 +65,15 @@ abstract class JoinSeedNodeSpec extends MultiNodeSpec(JoinSeedNodeMultiJvmSpec)
         cluster.joinSeedNodes(seedNodes)
       }
       awaitMembersUp(roles.size)
+
+      seedNodes.foreach { a =>
+        cluster.state.members.find(_.address == a).get.appVersion should ===(Version("1.0"))
+      }
+      List(address(ordinary1), address(ordinary2)).foreach { a =>
+        cluster.state.members.find(_.address == a).get.appVersion should ===(Version("2.0"))
+      }
+      cluster.state.hasMoreThanOneAppVersion should ===(true)
+
       enterBarrier("after-2")
     }
   }
diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/MBeanSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/MBeanSpec.scala
index e41faa3f90..7ee169f3e7 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/MBeanSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/MBeanSpec.scala
@@ -26,6 +26,7 @@ object MBeanMultiJvmSpec extends MultiNodeConfig {
   commonConfig(debugConfig(on = false).withFallback(ConfigFactory.parseString("""
     akka.cluster.jmx.enabled = on
     akka.cluster.roles = [testNode]
+    akka.cluster.app-version = "1.2.3"
     """)).withFallback(MultiNodeClusterSpec.clusterConfig))
 
 }
@@ -122,7 +123,8 @@ abstract class MBeanSpec extends MultiNodeSpec(MBeanMultiJvmSpec) with MultiNode
              |        "dc-default",
              |        "testNode"
              |      ],
-             |      "status": "Up"
+             |      "status": "Up",
+             |      "app-version": "1.2.3"
              |    },
              |    {
              |      "address": "${sortedNodes(1)}",
@@ -130,7 +132,8 @@ abstract class MBeanSpec extends MultiNodeSpec(MBeanMultiJvmSpec) with MultiNode
              |        "dc-default",
              |        "testNode"
              |      ],
-             |      "status": "Up"
+             |      "status": "Up",
+             |      "app-version": "1.2.3"
              |    },
              |    {
              |      "address": "${sortedNodes(2)}",
@@ -138,7 +141,8 @@ abstract class MBeanSpec extends MultiNodeSpec(MBeanMultiJvmSpec) with MultiNode
              |        "dc-default",
              |        "testNode"
              |      ],
-             |      "status": "Up"
+             |      "status": "Up",
+             |      "app-version": "1.2.3"
              |    },
              |    {
              |      "address": "${sortedNodes(3)}",
@@ -146,7 +150,8 @@ abstract class MBeanSpec extends MultiNodeSpec(MBeanMultiJvmSpec) with MultiNode
              |        "dc-default",
              |        "testNode"
              |      ],
-             |      "status": "Up"
+             |      "status": "Up",
+             |      "app-version": "1.2.3"
              |    }
              |  ],
              |  "self-address": "${address(first)}",
diff --git a/akka-cluster/src/test/scala/akka/cluster/ClusterConfigSpec.scala b/akka-cluster/src/test/scala/akka/cluster/ClusterConfigSpec.scala
index 718ce9ab69..cc0a2f078d 100644
--- a/akka-cluster/src/test/scala/akka/cluster/ClusterConfigSpec.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/ClusterConfigSpec.scala
@@ -15,6 +15,7 @@ import akka.dispatch.Dispatchers
 import akka.remote.PhiAccrualFailureDetector
 import akka.testkit.AkkaSpec
 import akka.util.Helpers.ConfigOps
+import akka.util.Version
 
 @silent
 class ClusterConfigSpec extends AkkaSpec {
@@ -48,6 +49,7 @@ class ClusterConfigSpec extends AkkaSpec {
       MinNrOfMembersOfRole should ===(Map.empty[String, Int])
       SelfDataCenter should ===("default")
       Roles should ===(Set(ClusterSettings.DcRolePrefix + "default"))
+      AppVersion should ===(Version.Zero)
       JmxEnabled should ===(true)
       UseDispatcher should ===(Dispatchers.InternalDispatcherId)
       GossipDifferentViewProbability should ===(0.8 +- 0.0001)
diff --git a/akka-cluster/src/test/scala/akka/cluster/ClusterHeartbeatSenderSpec.scala b/akka-cluster/src/test/scala/akka/cluster/ClusterHeartbeatSenderSpec.scala
index 5d23952f14..fbf82d8aa0 100644
--- a/akka-cluster/src/test/scala/akka/cluster/ClusterHeartbeatSenderSpec.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/ClusterHeartbeatSenderSpec.scala
@@ -9,6 +9,7 @@ import akka.cluster.ClusterEvent.{ CurrentClusterState, MemberUp }
 import akka.cluster.ClusterHeartbeatSender.Heartbeat
 import akka.cluster.ClusterHeartbeatSenderSpec.TestClusterHeartBeatSender
 import akka.testkit.{ AkkaSpec, ImplicitSender, TestProbe }
+import akka.util.Version
 
 object ClusterHeartbeatSenderSpec {
   class TestClusterHeartBeatSender(probe: TestProbe) extends ClusterHeartbeatSender {
@@ -34,7 +35,8 @@ class ClusterHeartbeatSenderSpec extends AkkaSpec("""
       val underTest = system.actorOf(Props(new TestClusterHeartBeatSender(probe)))
       underTest ! CurrentClusterState()
       underTest ! MemberUp(
-        Member(UniqueAddress(Address("akka", system.name), 1L), Set("dc-default")).copy(status = MemberStatus.Up))
+        Member(UniqueAddress(Address("akka", system.name), 1L), Set("dc-default"), Version.Zero)
+          .copy(status = MemberStatus.Up))
 
       probe.expectMsgType[Heartbeat].sequenceNr shouldEqual 1
       probe.expectMsgType[Heartbeat].sequenceNr shouldEqual 2
diff --git a/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala b/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala
index 512befffae..d67875ec0d 100644
--- a/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/ClusterSpec.scala
@@ -5,8 +5,8 @@
 package akka.cluster
 
 import java.lang.management.ManagementFactory
-import javax.management.ObjectName
 
+import javax.management.ObjectName
 import scala.concurrent.Await
 import scala.concurrent.duration._
 
@@ -27,6 +27,7 @@ import akka.stream.scaladsl.StreamRefs
 import akka.testkit.AkkaSpec
 import akka.testkit.ImplicitSender
 import akka.testkit.TestProbe
+import akka.util.Version
 
 object ClusterSpec {
   val config = """
@@ -36,6 +37,7 @@ object ClusterSpec {
       periodic-tasks-initial-delay = 120 seconds // turn off scheduled tasks
       publish-stats-interval = 0 s # always, when it happens
       failure-detector.implementation-class = akka.cluster.FailureDetectorPuppet
+      app-version = "1.2.3"
     }
     akka.actor.provider = "cluster"
     akka.remote.log-remote-lifecycle-events = off
@@ -101,6 +103,9 @@ class ClusterSpec extends AkkaSpec(ClusterSpec.config) with ImplicitSender {
       clusterView.self.address should ===(selfAddress)
       clusterView.members.map(_.address) should ===(Set(selfAddress))
       awaitAssert(clusterView.status should ===(MemberStatus.Up))
+      clusterView.self.appVersion should ===(Version("1.2.3"))
+      clusterView.members.find(_.address == selfAddress).get.appVersion should ===(Version("1.2.3"))
+      clusterView.state.hasMoreThanOneAppVersion should ===(false)
     }
 
     "reply with InitJoinAck for InitJoin after joining" in {
diff --git a/akka-cluster/src/test/scala/akka/cluster/CrossDcHeartbeatSenderSpec.scala b/akka-cluster/src/test/scala/akka/cluster/CrossDcHeartbeatSenderSpec.scala
index 032ef93816..9018a2a05e 100644
--- a/akka-cluster/src/test/scala/akka/cluster/CrossDcHeartbeatSenderSpec.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/CrossDcHeartbeatSenderSpec.scala
@@ -11,6 +11,7 @@ import akka.cluster.ClusterEvent.CurrentClusterState
 import akka.cluster.ClusterHeartbeatSender.Heartbeat
 import akka.cluster.CrossDcHeartbeatSenderSpec.TestCrossDcHeartbeatSender
 import akka.testkit.{ AkkaSpec, ImplicitSender, TestProbe }
+import akka.util.Version
 
 object CrossDcHeartbeatSenderSpec {
   class TestCrossDcHeartbeatSender(probe: TestProbe) extends CrossDcHeartbeatSender {
@@ -41,7 +42,8 @@ class CrossDcHeartbeatSenderSpec extends AkkaSpec("""
       underTest ! CurrentClusterState(
         members = SortedSet(
           Cluster(system).selfMember,
-          Member(UniqueAddress(Address("akka", system.name), 2L), Set("dc-dc2")).copy(status = MemberStatus.Up)))
+          Member(UniqueAddress(Address("akka", system.name), 2L), Set("dc-dc2"), Version.Zero)
+            .copy(status = MemberStatus.Up)))
 
       probe.expectMsgType[Heartbeat].sequenceNr shouldEqual 1
       probe.expectMsgType[Heartbeat].sequenceNr shouldEqual 2
diff --git a/akka-cluster/src/test/scala/akka/cluster/MemberOrderingSpec.scala b/akka-cluster/src/test/scala/akka/cluster/MemberOrderingSpec.scala
index 86abcd3651..f6e3fc1273 100644
--- a/akka-cluster/src/test/scala/akka/cluster/MemberOrderingSpec.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/MemberOrderingSpec.scala
@@ -10,7 +10,9 @@ import scala.util.Random
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
 
-import akka.actor.{ Address, AddressFromURIString }
+import akka.actor.Address
+import akka.actor.AddressFromURIString
+import akka.util.Version
 
 class MemberOrderingSpec extends AnyWordSpec with Matchers {
   import Member.addressOrdering
@@ -52,7 +54,7 @@ class MemberOrderingSpec extends AnyWordSpec with Matchers {
     "have stable equals and hashCode" in {
       val address = Address("akka", "sys1", "host1", 9000)
       val m1 = m(address, Joining)
-      val m11 = Member(UniqueAddress(address, -3L), Set.empty)
+      val m11 = Member(UniqueAddress(address, -3L), Set.empty, Version.Zero)
       val m2 = m1.copy(status = Up)
       val m22 = m11.copy(status = Up)
       val m3 = m(address.copy(port = Some(10000)), Up)
@@ -83,7 +85,7 @@ class MemberOrderingSpec extends AnyWordSpec with Matchers {
 
       // different uid
       val a = m(address1, Joining)
-      val b = Member(UniqueAddress(address1, -3L), Set.empty)
+      val b = Member(UniqueAddress(address1, -3L), Set.empty, Version.Zero)
       Member.ordering.compare(a, b) should ===(1)
       Member.ordering.compare(b, a) should ===(-1)
 
diff --git a/akka-cluster/src/test/scala/akka/cluster/TestMember.scala b/akka-cluster/src/test/scala/akka/cluster/TestMember.scala
index bf9735ffc0..0334560e7f 100644
--- a/akka-cluster/src/test/scala/akka/cluster/TestMember.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/TestMember.scala
@@ -5,6 +5,7 @@
 package akka.cluster
 
 import akka.actor.Address
+import akka.util.Version
 
 object TestMember {
   def apply(address: Address, status: MemberStatus): Member =
@@ -18,14 +19,16 @@ object TestMember {
       status: MemberStatus,
       roles: Set[String],
       dataCenter: ClusterSettings.DataCenter = ClusterSettings.DefaultDataCenter,
-      upNumber: Int = Int.MaxValue): Member =
-    withUniqueAddress(UniqueAddress(address, 0L), status, roles, dataCenter, upNumber)
+      upNumber: Int = Int.MaxValue,
+      appVersion: Version = Version.Zero): Member =
+    withUniqueAddress(UniqueAddress(address, 0L), status, roles, dataCenter, upNumber, appVersion)
 
   def withUniqueAddress(
       uniqueAddress: UniqueAddress,
       status: MemberStatus,
       roles: Set[String],
       dataCenter: ClusterSettings.DataCenter,
-      upNumber: Int = Int.MaxValue): Member =
-    new Member(uniqueAddress, upNumber, status, roles + (ClusterSettings.DcRolePrefix + dataCenter))
+      upNumber: Int = Int.MaxValue,
+      appVersion: Version = Version.Zero): Member =
+    new Member(uniqueAddress, upNumber, status, roles + (ClusterSettings.DcRolePrefix + dataCenter), appVersion)
 }
diff --git a/akka-cluster/src/test/scala/akka/cluster/protobuf/ClusterMessageSerializerSpec.scala b/akka-cluster/src/test/scala/akka/cluster/protobuf/ClusterMessageSerializerSpec.scala
index a9edc15f1b..afd57aa471 100644
--- a/akka-cluster/src/test/scala/akka/cluster/protobuf/ClusterMessageSerializerSpec.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/protobuf/ClusterMessageSerializerSpec.scala
@@ -5,6 +5,7 @@
 package akka.cluster.protobuf
 
 import collection.immutable.SortedSet
+
 import com.github.ghik.silencer.silent
 import com.typesafe.config.ConfigFactory
 
@@ -14,6 +15,7 @@ import akka.cluster.InternalClusterAction.CompatibleConfig
 import akka.cluster.routing.{ ClusterRouterPool, ClusterRouterPoolSettings }
 import akka.routing.RoundRobinPool
 import akka.testkit.AkkaSpec
+import akka.util.Version
 
 @silent
 class ClusterMessageSerializerSpec extends AkkaSpec("akka.actor.provider = cluster") {
@@ -48,16 +50,31 @@ class ClusterMessageSerializerSpec extends AkkaSpec("akka.actor.provider = clust
         env2.from should ===(env.from)
         env2.to should ===(env.to)
         env2.gossip should ===(env.gossip)
+        env.gossip.members.foreach { m1 =>
+          val m2 = env.gossip.members.find(_.uniqueAddress == m1.uniqueAddress).get
+          checkSameMember(m1, m2)
+        }
       case (_, ref) =>
         ref should ===(obj)
     }
   }
 
+  private def checkSameMember(m1: Member, m2: Member): Unit = {
+    m1.uniqueAddress should ===(m2.uniqueAddress)
+    m1.status should ===(m2.status)
+    m1.appVersion should ===(m2.appVersion)
+    m1.dataCenter should ===(m2.dataCenter)
+    m1.roles should ===(m2.roles)
+    m1.upNumber should ===(m2.upNumber)
+  }
+
   import MemberStatus._
 
-  val a1 = TestMember(Address("akka", "sys", "a", 2552), Joining, Set.empty[String])
-  val b1 = TestMember(Address("akka", "sys", "b", 2552), Up, Set("r1"))
-  val c1 = TestMember(Address("akka", "sys", "c", 2552), Leaving, Set.empty[String], "foo")
+  val a1 =
+    TestMember(Address("akka", "sys", "a", 2552), Joining, Set.empty[String], appVersion = Version("1.0.0"))
+  val b1 = TestMember(Address("akka", "sys", "b", 2552), Up, Set("r1"), appVersion = Version("1.1.0"))
+  val c1 =
+    TestMember(Address("akka", "sys", "c", 2552), Leaving, Set.empty[String], "foo", appVersion = Version("1.1.0"))
   val d1 = TestMember(Address("akka", "sys", "d", 2552), Exiting, Set("r1"), "foo")
   val e1 = TestMember(Address("akka", "sys", "e", 2552), Down, Set("r3"))
   val f1 = TestMember(Address("akka", "sys", "f", 2552), Removed, Set("r3"), "foo")
@@ -69,7 +86,8 @@ class ClusterMessageSerializerSpec extends AkkaSpec("akka.actor.provider = clust
       val uniqueAddress = UniqueAddress(address, 17L)
       val address2 = Address("akka", "system", "other.host.org", 4711)
       val uniqueAddress2 = UniqueAddress(address2, 18L)
-      checkSerialization(InternalClusterAction.Join(uniqueAddress, Set("foo", "bar", "dc-A")))
+      checkSerialization(InternalClusterAction.Join(uniqueAddress, Set("foo", "bar", "dc-A"), Version.Zero))
+      checkSerialization(InternalClusterAction.Join(uniqueAddress, Set("dc-A"), Version("1.2.3")))
       checkSerialization(ClusterUserAction.Leave(address))
       checkSerialization(ClusterUserAction.Down(address))
       checkSerialization(InternalClusterAction.InitJoin(ConfigFactory.empty))
@@ -110,7 +128,7 @@ class ClusterMessageSerializerSpec extends AkkaSpec("akka.actor.provider = clust
       val address2 = Address("akka", "system", "other.host.org", 4711)
       val uniqueAddress2 = UniqueAddress(address2, 18L)
       checkDeserializationWithManifest(
-        InternalClusterAction.Join(uniqueAddress, Set("foo", "bar", "dc-A")),
+        InternalClusterAction.Join(uniqueAddress, Set("foo", "bar", "dc-A"), Version.Zero),
         ClusterMessageSerializer.OldJoinManifest)
       checkDeserializationWithManifest(ClusterUserAction.Leave(address), ClusterMessageSerializer.LeaveManifest)
       checkDeserializationWithManifest(ClusterUserAction.Down(address), ClusterMessageSerializer.DownManifest)
@@ -156,7 +174,7 @@ class ClusterMessageSerializerSpec extends AkkaSpec("akka.actor.provider = clust
     }
 
     "add a default data center role to internal join action if none is present" in {
-      val join = roundtrip(InternalClusterAction.Join(a1.uniqueAddress, Set()))
+      val join = roundtrip(InternalClusterAction.Join(a1.uniqueAddress, Set(), Version.Zero))
       join.roles should be(Set(ClusterSettings.DcRolePrefix + "default"))
     }
   }
diff --git a/akka-cluster/src/test/scala/akka/cluster/sbr/TestAddresses.scala b/akka-cluster/src/test/scala/akka/cluster/sbr/TestAddresses.scala
index 3ac75dc830..e4b6b6f819 100644
--- a/akka-cluster/src/test/scala/akka/cluster/sbr/TestAddresses.scala
+++ b/akka-cluster/src/test/scala/akka/cluster/sbr/TestAddresses.scala
@@ -11,6 +11,7 @@ import akka.cluster.MemberStatus
 import akka.cluster.MemberStatus.Up
 import akka.cluster.MemberStatus.WeaklyUp
 import akka.cluster.UniqueAddress
+import akka.util.Version
 
 /**
  * Needed since the Member constructor is akka private
@@ -22,30 +23,45 @@ object TestAddresses {
   private def defaultDcRole = dcRole(defaultDataCenter)
 
   val addressA = Address("akka.tcp", "sys", "a", 2552)
-  val memberA = new Member(UniqueAddress(addressA, 0L), 5, Up, Set("role3", defaultDcRole))
+  val memberA = new Member(UniqueAddress(addressA, 0L), 5, Up, Set("role3", defaultDcRole), Version.Zero)
   val memberB =
-    new Member(UniqueAddress(addressA.copy(host = Some("b")), 0L), 4, Up, Set("role1", "role3", defaultDcRole))
-  val memberC = new Member(UniqueAddress(addressA.copy(host = Some("c")), 0L), 3, Up, Set("role2", defaultDcRole))
+    new Member(
+      UniqueAddress(addressA.copy(host = Some("b")), 0L),
+      4,
+      Up,
+      Set("role1", "role3", defaultDcRole),
+      Version.Zero)
+  val memberC =
+    new Member(UniqueAddress(addressA.copy(host = Some("c")), 0L), 3, Up, Set("role2", defaultDcRole), Version.Zero)
   val memberD =
-    new Member(UniqueAddress(addressA.copy(host = Some("d")), 0L), 2, Up, Set("role1", "role2", "role3", defaultDcRole))
-  val memberE = new Member(UniqueAddress(addressA.copy(host = Some("e")), 0L), 1, Up, Set(defaultDcRole))
-  val memberF = new Member(UniqueAddress(addressA.copy(host = Some("f")), 0L), 5, Up, Set(defaultDcRole))
-  val memberG = new Member(UniqueAddress(addressA.copy(host = Some("g")), 0L), 6, Up, Set(defaultDcRole))
+    new Member(
+      UniqueAddress(addressA.copy(host = Some("d")), 0L),
+      2,
+      Up,
+      Set("role1", "role2", "role3", defaultDcRole),
+      Version.Zero)
+  val memberE =
+    new Member(UniqueAddress(addressA.copy(host = Some("e")), 0L), 1, Up, Set(defaultDcRole), Version.Zero)
+  val memberF =
+    new Member(UniqueAddress(addressA.copy(host = Some("f")), 0L), 5, Up, Set(defaultDcRole), Version.Zero)
+  val memberG =
+    new Member(UniqueAddress(addressA.copy(host = Some("g")), 0L), 6, Up, Set(defaultDcRole), Version.Zero)
 
-  val memberAWeaklyUp = new Member(memberA.uniqueAddress, Int.MaxValue, WeaklyUp, memberA.roles)
-  val memberBWeaklyUp = new Member(memberB.uniqueAddress, Int.MaxValue, WeaklyUp, memberB.roles)
+  val memberAWeaklyUp = new Member(memberA.uniqueAddress, Int.MaxValue, WeaklyUp, memberA.roles, Version.Zero)
+  val memberBWeaklyUp = new Member(memberB.uniqueAddress, Int.MaxValue, WeaklyUp, memberB.roles, Version.Zero)
 
   def dcMember(dc: ClusterSettings.DataCenter, m: Member): Member =
     new Member(
       m.uniqueAddress,
       m.upNumber,
       m.status,
-      m.roles.filterNot(_.startsWith(ClusterSettings.DcRolePrefix)) + dcRole(dc))
+      m.roles.filterNot(_.startsWith(ClusterSettings.DcRolePrefix)) + dcRole(dc),
+      Version.Zero)
 
   def dataCenter(dc: ClusterSettings.DataCenter, members: Member*): Set[Member] =
     members.toSet[Member].map(m => dcMember(dc, m))
 
-  def joining(m: Member): Member = Member(m.uniqueAddress, m.roles)
+  def joining(m: Member): Member = Member(m.uniqueAddress, m.roles, Version.Zero)
 
   def leaving(m: Member): Member = m.copy(MemberStatus.Leaving)