Verify removal and add of new node incarnation in multi-dc, #23585

akka-cluster-tools/src/multi-jvm/scala/akka/cluster/singleton/MultiDcSingletonManagerSpec.scala

@@ -21,7 +21,7 @@ object MultiDcSingletonManagerSpec extends MultiNodeConfig {
   val third = role("third")
 
   commonConfig(ConfigFactory.parseString("""
-    akka.loglevel = INFO
+    akka.loglevel = DEBUG
     akka.actor.provider = "cluster"
     akka.actor.serialize-creators = off
     akka.remote.log-remote-lifecycle-events = off"""))
@@ -94,7 +94,7 @@ abstract class MultiDcSingletonManagerSpec extends MultiNodeSpec(MultiDcSingleto
       enterBarrier("managers-started")
 
       proxy ! MultiDcSingleton.Ping
-      val pong = expectMsgType[MultiDcSingleton.Pong](10.seconds)
+      val pong = expectMsgType[MultiDcSingleton.Pong](20.seconds)
 
       enterBarrier("pongs-received")
 

akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala

@@ -886,10 +886,10 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
         else winningGossip seen selfUniqueAddress)
       assertLatestGossip()
 
-      // for all new joining nodes we remove them from the failure detector
+      // for all new nodes we remove them from the failure detector
       latestGossip.members foreach {
         node ⇒
-          if (node.status == Joining && !localGossip.members(node)) {
+          if (!localGossip.members(node)) {
             failureDetector.remove(node.address)
             crossDcFailureDetector.remove(node.address)
           }

akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala

@@ -454,7 +454,7 @@ object ClusterEvent {
       val removedMembers = oldGossip.members diff newGossip.members
       val removedEvents = removedMembers.map(m ⇒ MemberRemoved(m.copy(status = Removed), m.status))
 
-      (new VectorBuilder[MemberEvent]() ++= memberEvents ++= removedEvents).result()
+      (new VectorBuilder[MemberEvent]() ++= removedEvents ++= memberEvents).result()
     }
 
   /**

akka-cluster/src/main/scala/akka/cluster/MembershipState.scala

@@ -130,7 +130,8 @@ import scala.util.Random
     }
 
   def dcMembers: SortedSet[Member] =
-    members.filter(_.dataCenter == selfDc)
+    if (latestGossip.isMultiDc) members.filter(_.dataCenter == selfDc)
+    else members
 
   def isLeader(node: UniqueAddress): Boolean =
     leader.contains(node)
@@ -159,17 +160,14 @@ import scala.util.Random
   def isInSameDc(node: UniqueAddress): Boolean =
     node == selfUniqueAddress || latestGossip.member(node).dataCenter == selfDc
 
-  def membersInSameDc: immutable.SortedSet[Member] =
-    members.filter(_.dataCenter == selfDc)
-
   def validNodeForGossip(node: UniqueAddress): Boolean =
     node != selfUniqueAddress &&
       ((isInSameDc(node) && isReachableExcludingDownedObservers(node)) ||
         // if cross DC we need to check pairwise unreachable observation
         overview.reachability.isReachable(selfUniqueAddress, node))
 
   def youngestMember: Member = {
-    val mbrs = membersInSameDc
+    val mbrs = dcMembers
     require(mbrs.nonEmpty, "No youngest when no members")
     mbrs.maxBy(m ⇒ if (m.upNumber == Int.MaxValue) 0 else m.upNumber)
   }

akka-cluster/src/multi-jvm/scala/akka/cluster/MultiDcSplitBrainSpec.scala

@@ -3,20 +3,27 @@
  */
 package akka.cluster
 
-import akka.cluster.ClusterEvent.{ CurrentClusterState, DataCenterReachabilityEvent, ReachableDataCenter, UnreachableDataCenter }
+import akka.cluster.ClusterEvent._
 import akka.remote.testconductor.RoleName
 import akka.remote.testkit.{ MultiNodeConfig, MultiNodeSpec }
 import akka.remote.transport.ThrottlerTransportAdapter.Direction
 import akka.testkit.TestProbe
 import com.typesafe.config.ConfigFactory
 
 import scala.concurrent.duration._
+import akka.actor.ActorSystem
+import akka.actor.Props
+import akka.actor.Actor
+import akka.actor.Deploy
+import akka.actor.RootActorPath
+import scala.concurrent.Await
 
 object MultiDcSplitBrainMultiJvmSpec extends MultiNodeConfig {
   val first = role("first")
   val second = role("second")
   val third = role("third")
   val fourth = role("fourth")
+  val fifth = role("fifth")
 
   commonConfig(ConfigFactory.parseString(
     """
@@ -30,14 +37,19 @@ object MultiDcSplitBrainMultiJvmSpec extends MultiNodeConfig {
           heartbeat-interval = 1s
         }
       }
+      akka.cluster {
+        gossip-interval                     = 1s
+        leader-actions-interval             = 1s
+        auto-down-unreachable-after = 1s
+      }
     """).withFallback(MultiNodeClusterSpec.clusterConfig))
 
   nodeConfig(first, second)(ConfigFactory.parseString(
     """
       akka.cluster.multi-data-center.self-data-center = "dc1"
     """))
 
-  nodeConfig(third, fourth)(ConfigFactory.parseString(
+  nodeConfig(third, fourth, fifth)(ConfigFactory.parseString(
     """
       akka.cluster.multi-data-center.self-data-center = "dc2"
     """))
@@ -49,6 +61,7 @@ class MultiDcSplitBrainMultiJvmNode1 extends MultiDcSplitBrainSpec
 class MultiDcSplitBrainMultiJvmNode2 extends MultiDcSplitBrainSpec
 class MultiDcSplitBrainMultiJvmNode3 extends MultiDcSplitBrainSpec
 class MultiDcSplitBrainMultiJvmNode4 extends MultiDcSplitBrainSpec
+class MultiDcSplitBrainMultiJvmNode5 extends MultiDcSplitBrainSpec
 
 abstract class MultiDcSplitBrainSpec
   extends MultiNodeSpec(MultiDcSplitBrainMultiJvmSpec)
@@ -57,7 +70,7 @@ abstract class MultiDcSplitBrainSpec
   import MultiDcSplitBrainMultiJvmSpec._
 
   val dc1 = List(first, second)
-  val dc2 = List(third, fourth)
+  val dc2 = List(third, fourth, fifth)
   var barrierCounter = 0
 
   def splitDataCenters(notMembers: Set[RoleName]): Unit = {
@@ -136,7 +149,7 @@ abstract class MultiDcSplitBrainSpec
 
     "be able to have a data center member join while there is inter data center split" in within(20.seconds) {
       // introduce a split between data centers
-      splitDataCenters(notMembers = Set(fourth))
+      splitDataCenters(notMembers = Set(fourth, fifth))
 
       runOn(fourth) {
         cluster.join(third)
@@ -152,7 +165,7 @@ abstract class MultiDcSplitBrainSpec
       }
       enterBarrier("dc2-join-completed")
 
-      unsplitDataCenters(notMembers = Set.empty)
+      unsplitDataCenters(notMembers = Set(fifth))
 
       runOn(dc1: _*) {
         awaitAssert(clusterView.members.collect {
@@ -164,7 +177,7 @@ abstract class MultiDcSplitBrainSpec
     }
 
     "be able to have data center member leave while there is inter data center split" in within(20.seconds) {
-      splitDataCenters(notMembers = Set.empty)
+      splitDataCenters(notMembers = Set(fifth))
 
       runOn(fourth) {
         cluster.leave(fourth)
@@ -175,13 +188,104 @@ abstract class MultiDcSplitBrainSpec
       }
       enterBarrier("node-4-left")
 
-      unsplitDataCenters(notMembers = Set(fourth))
+      unsplitDataCenters(notMembers = Set(fourth, fifth))
 
       runOn(first, second) {
         awaitAssert(clusterView.members.filter(_.address == address(fourth)) should ===(Set.empty))
       }
       enterBarrier("inter-data-center-split-2-done")
     }
 
+    "be able to have data center member restart (same host:port) while there is inter data center split" in within(40.seconds) {
+      val subscribeProbe = TestProbe()
+      runOn(first, second, third, fifth) {
+        Cluster(system).subscribe(subscribeProbe.ref, InitialStateAsSnapshot, classOf[MemberUp], classOf[MemberRemoved])
+        subscribeProbe.expectMsgType[CurrentClusterState]
+      }
+      enterBarrier("subscribed")
+      runOn(fifth) {
+        Cluster(system).join(third)
+      }
+      var fifthOriginalUniqueAddress: Option[UniqueAddress] = None
+      runOn(first, second, third, fifth) {
+        awaitAssert(clusterView.members.collect {
+          case m if m.dataCenter == "dc2" && m.status == MemberStatus.Up ⇒ m.address
+        } should ===(Set(address(third), address(fifth))))
+        fifthOriginalUniqueAddress = clusterView.members.collectFirst { case m if m.address == address(fifth) ⇒ m.uniqueAddress }
+      }
+      enterBarrier("fifth-joined")
+
+      splitDataCenters(notMembers = Set(fourth))
+
+      runOn(fifth) {
+        Cluster(system).shutdown()
+      }
+      runOn(third) {
+        awaitAssert(clusterView.members.collect {
+          case m if m.dataCenter == "dc2" ⇒ m.address
+        } should ===(Set(address(third))))
+      }
+      enterBarrier("fifth-removed")
+
+      runOn(fifth) {
+        // we can't use any multi-jvm test facilities on fifth after this, because have to shutdown
+        // actor system to be able to start new with same port
+        val thirdAddress = address(third)
+        enterBarrier("fifth-waiting-for-termination")
+        Await.ready(system.whenTerminated, remaining)
+
+        val restartedSystem = ActorSystem(
+          system.name,
+          ConfigFactory.parseString(s"""
+            akka.remote.netty.tcp.port = ${Cluster(system).selfAddress.port.get}
+            akka.remote.artery.canonical.port = ${Cluster(system).selfAddress.port.get}
+            akka.coordinated-shutdown.terminate-actor-system = on
+            """).withFallback(system.settings.config))
+        Cluster(restartedSystem).join(thirdAddress)
+        Await.ready(restartedSystem.whenTerminated, remaining)
+      }
+
+      // no multi-jvm test facilities on fifth after this
+      val remainingRoles = roles.filterNot(_ == fifth)
+
+      runOn(remainingRoles: _*) {
+        enterBarrier("fifth-waiting-for-termination")
+      }
+
+      runOn(first) {
+        for (dc1Node ← dc1; dc2Node ← dc2) {
+          testConductor.passThrough(dc1Node, dc2Node, Direction.Both).await
+        }
+        testConductor.shutdown(fifth)
+      }
+      runOn(remainingRoles: _*) {
+        enterBarrier("fifth-restarted")
+      }
+
+      runOn(first, second, third) {
+        awaitAssert(clusterView.members.collectFirst {
+          case m if m.dataCenter == "dc2" && m.address == fifthOriginalUniqueAddress.get.address ⇒ m.uniqueAddress
+        } should not be (fifthOriginalUniqueAddress)) // different uid
+        subscribeProbe.expectMsgType[MemberUp].member.uniqueAddress should ===(fifthOriginalUniqueAddress.get)
+        subscribeProbe.expectMsgType[MemberRemoved].member.uniqueAddress should ===(fifthOriginalUniqueAddress.get)
+        subscribeProbe.expectMsgType[MemberUp].member.address should ===(fifthOriginalUniqueAddress.get.address)
+      }
+      runOn(remainingRoles: _*) {
+        enterBarrier("fifth-re-joined")
+      }
+      runOn(first) {
+        // to shutdown the restartedSystem on fifth
+        Cluster(system).leave(fifthOriginalUniqueAddress.get.address)
+      }
+      runOn(first, second, third) {
+        awaitAssert(clusterView.members.map(_.address) should ===(Set(address(first), address(second), address(third))))
+      }
+      runOn(remainingRoles: _*) {
+        Thread.sleep(5000) // FIXME remove
+        enterBarrier("restarted-fifth-removed")
+      }
+
+    }
+
   }
 }

akka-cluster/src/test/scala/akka/cluster/ClusterDomainEventSpec.scala

@@ -180,6 +180,20 @@ class ClusterDomainEventSpec extends WordSpec with Matchers {
       diffSeen(state(g1), state(g2)) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address))))
     }
 
+    "be produced for removed and rejoined member in another data center" in {
+      val bUpDc2 = TestMember(bUp.address, Up, bRoles, dataCenter = "dc2")
+      val bUpDc2Removed = TestMember(bUpDc2.address, Removed, bRoles, dataCenter = "dc2")
+      val bUpDc2Restarted = TestMember.withUniqueAddress(UniqueAddress(bUpDc2.address, 2L), Up, bRoles, dataCenter = "dc2")
+      val g1 = Gossip(members = SortedSet(aUp, bUpDc2))
+      val g2 = g1
+        .remove(bUpDc2.uniqueAddress, System.currentTimeMillis()) // adds tombstone
+        .copy(members = SortedSet(aUp, bUpDc2Restarted))
+        .merge(g1)
+
+      diffMemberEvents(state(g1), state(g2)) should ===(Seq(
+        MemberRemoved(bUpDc2Removed, Up), MemberUp(bUpDc2Restarted)))
+    }
+
     "be produced for convergence changes" in {
       val g1 = Gossip(members = SortedSet(aUp, bUp, eJoining)).seen(aUp.uniqueAddress).seen(bUp.uniqueAddress).seen(eJoining.uniqueAddress)
       val g2 = Gossip(members = SortedSet(aUp, bUp, eJoining)).seen(aUp.uniqueAddress).seen(bUp.uniqueAddress)

akka-cluster/src/test/scala/akka/cluster/GossipSpec.scala

@@ -33,6 +33,8 @@ class GossipSpec extends WordSpec with Matchers {
   val dc2c1 = TestMember(Address("akka.tcp", "sys", "c", 2552), Up, Set.empty, dataCenter = "dc2")
   val dc2d1 = TestMember(Address("akka.tcp", "sys", "d", 2552), Up, Set.empty, dataCenter = "dc2")
   val dc2d2 = TestMember(dc2d1.address, status = Down, roles = Set.empty, dataCenter = dc2d1.dataCenter)
+  // restarted with another uid
+  val dc2d3 = TestMember.withUniqueAddress(UniqueAddress(dc2d1.address, longUid = 3L), Up, Set.empty, dataCenter = "dc2")
 
   private def state(g: Gossip, selfMember: Member = a1): MembershipState =
     MembershipState(g, selfMember.uniqueAddress, selfMember.dataCenter, crossDcConnections = 5)
@@ -387,6 +389,36 @@ class GossipSpec extends WordSpec with Matchers {
       merged1.version.versions.keys should not contain (VectorClock.Node(vclockName(dc2d1.uniqueAddress)))
     }
 
+    "replace member when removed and rejoined in another data center" in {
+      // dc1 does not know removal and rejoin of new incarnation
+      val gdc1 = Gossip(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d1))
+        .seen(dc1b1.uniqueAddress)
+        .seen(dc2c1.uniqueAddress)
+        .:+(VectorClock.Node(vclockName(dc2d1.uniqueAddress))) // just to make sure these are also pruned
+
+      // dc2 has removed the dc2d1 node, and then same host:port is restarted and joins again, without dc1 knowning
+      val gdc2 = Gossip(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d3))
+        .seen(dc1a1.uniqueAddress)
+        .remove(dc2d1.uniqueAddress, System.currentTimeMillis())
+        .copy(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d3))
+
+      gdc2.tombstones.keys should contain(dc2d1.uniqueAddress)
+      gdc2.members.map(_.uniqueAddress) should not contain (dc2d1.uniqueAddress)
+      gdc2.members.map(_.uniqueAddress) should contain(dc2d3.uniqueAddress)
+
+      // when we merge the two, it should replace the old with new
+      val merged1 = gdc2 merge gdc1
+      merged1.members should ===(SortedSet(dc1a1, dc1b1, dc2c1, dc2d3))
+      merged1.members.map(_.uniqueAddress) should not contain (dc2d1.uniqueAddress)
+      merged1.members.map(_.uniqueAddress) should contain(dc2d3.uniqueAddress)
+
+      merged1.tombstones.keys should contain(dc2d1.uniqueAddress)
+      merged1.tombstones.keys should not contain (dc2d3.uniqueAddress)
+      merged1.overview.reachability.records.filter(r ⇒ r.subject == dc2d1.uniqueAddress || r.observer == dc2d1.uniqueAddress) should be(empty)
+      merged1.overview.reachability.versions.keys should not contain (dc2d1.uniqueAddress)
+      merged1.version.versions.keys should not contain (VectorClock.Node(vclockName(dc2d1.uniqueAddress)))
+    }
+
     "correctly prune vector clocks based on tombstones when merging" in {
       val gdc1 = Gossip(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d1))
         .:+(VectorClock.Node(vclockName(dc1a1.uniqueAddress)))

akka-cluster/src/test/scala/akka/cluster/TestMember.scala

@@ -10,5 +10,8 @@ object TestMember {
     apply(address, status, Set.empty)
 
   def apply(address: Address, status: MemberStatus, roles: Set[String], dataCenter: ClusterSettings.DataCenter = ClusterSettings.DefaultDataCenter): Member =
-    new Member(UniqueAddress(address, 0L), Int.MaxValue, status, roles + (ClusterSettings.DcRolePrefix + dataCenter))
+    withUniqueAddress(UniqueAddress(address, 0L), status, roles, dataCenter)
+
+  def withUniqueAddress(uniqueAddress: UniqueAddress, status: MemberStatus, roles: Set[String], dataCenter: ClusterSettings.DataCenter): Member =
+    new Member(uniqueAddress, Int.MaxValue, status, roles + (ClusterSettings.DcRolePrefix + dataCenter))
 }