raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Change member prio for Joining/Up, see #3239

Browse source 
* Sometimes caused duplicate MemberUp events, after conflicting gossips
This commit is contained in:
Patrik Nordwall 2013-04-19 07:54:23 +02:00
parent 3569886bbe
commit a3df775931
6 changed files with 25 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

23
akka-cluster/src/main/scala/akka/cluster/Member.scala
View file

@ -127,18 +127,17 @@ object Member {
* Picks the Member with the highest "priority" MemberStatus. * Picks the Member with the highest "priority" MemberStatus.
*/ */
def highestPriorityOf(m1: Member, m2: Member): Member = (m1.status, m2.status) match { def highestPriorityOf(m1: Member, m2: Member): Member = (m1.status, m2.status) match {
case (Removed, _) ⇒ m1 case (Removed, _) ⇒ m1
case (_, Removed) ⇒ m2 case (_, Removed) ⇒ m2
case (Down, _) ⇒ m1 case (Down, _) ⇒ m1
case (_, Down) ⇒ m2 case (_, Down) ⇒ m2
case (Exiting, _) ⇒ m1 case (Exiting, _) ⇒ m1
case (_, Exiting) ⇒ m2 case (_, Exiting) ⇒ m2
case (Leaving, _) ⇒ m1 case (Leaving, _) ⇒ m1
case (_, Leaving) ⇒ m2 case (_, Leaving) ⇒ m2
case (Up, Joining) ⇒ m2 case (Joining, _) ⇒ m2
case (Joining, Up) ⇒ m1 case (_, Joining) ⇒ m1
case (Joining, Joining) ⇒ m1 case (Up, Up) ⇒ m1
case (Up, Up) ⇒ m1
} }
} }

12
akka-cluster/src/test/scala/akka/cluster/GossipSpec.scala
View file

@ -39,11 +39,11 @@ class GossipSpec extends WordSpec with MustMatchers {
val merged1 = g1 merge g2 val merged1 = g1 merge g2
merged1.members must be(SortedSet(a2, c1, e1)) merged1.members must be(SortedSet(a2, c1, e1))
merged1.members.toSeq.map(_.status) must be(Seq(Joining, Leaving, Joining)) merged1.members.toSeq.map(_.status) must be(Seq(Up, Leaving, Up))
val merged2 = g2 merge g1 val merged2 = g2 merge g1
merged2.members must be(SortedSet(a2, c1, e1)) merged2.members must be(SortedSet(a2, c1, e1))
merged2.members.toSeq.map(_.status) must be(Seq(Joining, Leaving, Joining)) merged2.members.toSeq.map(_.status) must be(Seq(Up, Leaving, Up))
} }
@ -53,11 +53,11 @@ class GossipSpec extends WordSpec with MustMatchers {
val merged1 = g1 merge g2 val merged1 = g1 merge g2
merged1.overview.unreachable must be(Set(a2, b2, c1, d2)) merged1.overview.unreachable must be(Set(a2, b2, c1, d2))
merged1.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Joining, Removed, Leaving, Removed)) merged1.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Up, Removed, Leaving, Removed))
val merged2 = g2 merge g1 val merged2 = g2 merge g1
merged2.overview.unreachable must be(Set(a2, b2, c1, d2)) merged2.overview.unreachable must be(Set(a2, b2, c1, d2))
merged2.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Joining, Removed, Leaving, Removed)) merged2.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Up, Removed, Leaving, Removed))
} }
@ -67,13 +67,13 @@ class GossipSpec extends WordSpec with MustMatchers {
val merged1 = g1 merge g2 val merged1 = g1 merge g2
merged1.members must be(SortedSet(a2)) merged1.members must be(SortedSet(a2))
merged1.members.toSeq.map(_.status) must be(Seq(Joining)) merged1.members.toSeq.map(_.status) must be(Seq(Up))
merged1.overview.unreachable must be(Set(b2, c1, d2)) merged1.overview.unreachable must be(Set(b2, c1, d2))
merged1.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Removed, Leaving, Removed)) merged1.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Removed, Leaving, Removed))
val merged2 = g2 merge g1 val merged2 = g2 merge g1
merged2.members must be(SortedSet(a2)) merged2.members must be(SortedSet(a2))
merged2.members.toSeq.map(_.status) must be(Seq(Joining)) merged2.members.toSeq.map(_.status) must be(Seq(Up))
merged2.overview.unreachable must be(Set(b2, c1, d2)) merged2.overview.unreachable must be(Set(b2, c1, d2))
merged2.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Removed, Leaving, Removed)) merged2.overview.unreachable.toSeq.sorted.map(_.status) must be(Seq(Removed, Leaving, Removed))

10
akka-samples/akka-sample-cluster/src/multi-jvm/scala/sample/cluster/stats/StatsSampleSingleMasterSpec.scala
View file

@ -80,14 +80,8 @@ abstract class StatsSampleSingleMasterSpec extends MultiNodeSpec(StatsSampleSing
Cluster(system) join firstAddress Cluster(system) join firstAddress
// FIXME ticket 3239 duplicate MemberUp events, it should be possible to use receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (
// receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (Set(firstAddress, secondAddress, thirdAddress)) Set(firstAddress, secondAddress, thirdAddress))
import akka.actor.Address
@scala.annotation.tailrec def awaitMembersUp(expected: Set[Address], got: Set[Address] = Set.empty): Unit = {
val members = got + expectMsgType[MemberUp].member.address
if (members != expected) awaitMembersUp(expected, members)
}
awaitMembersUp(Set(firstAddress, secondAddress, thirdAddress))
Cluster(system).unsubscribe(testActor) Cluster(system).unsubscribe(testActor)

10
akka-samples/akka-sample-cluster/src/multi-jvm/scala/sample/cluster/stats/StatsSampleSpec.scala
View file

@ -97,14 +97,8 @@ abstract class StatsSampleSpec extends MultiNodeSpec(StatsSampleSpecConfig)
system.actorOf(Props[StatsWorker], "statsWorker") system.actorOf(Props[StatsWorker], "statsWorker")
system.actorOf(Props[StatsService], "statsService") system.actorOf(Props[StatsService], "statsService")
// FIXME ticket 3239 duplicate MemberUp events, it should be possible to use receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (
// receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (Set(firstAddress, secondAddress, thirdAddress)) Set(firstAddress, secondAddress, thirdAddress))
import akka.actor.Address
@scala.annotation.tailrec def awaitMembersUp(expected: Set[Address], got: Set[Address] = Set.empty): Unit = {
val members = got + expectMsgType[MemberUp].member.address
if (members != expected) awaitMembersUp(expected, members)
}
awaitMembersUp(Set(firstAddress, secondAddress, thirdAddress))
Cluster(system).unsubscribe(testActor) Cluster(system).unsubscribe(testActor)

10
akka-samples/akka-sample-cluster/src/multi-jvm/scala/sample/cluster/stats/japi/StatsSampleJapiSpec.scala
View file

@ -82,14 +82,8 @@ abstract class StatsSampleJapiSpec extends MultiNodeSpec(StatsSampleJapiSpecConf
system.actorOf(Props[StatsWorker], "statsWorker") system.actorOf(Props[StatsWorker], "statsWorker")
system.actorOf(Props[StatsService], "statsService") system.actorOf(Props[StatsService], "statsService")
// FIXME ticket 3239 duplicate MemberUp events, it should be possible to use receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (
// receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (Set(firstAddress, secondAddress, thirdAddress)) Set(firstAddress, secondAddress, thirdAddress))
import akka.actor.Address
@scala.annotation.tailrec def awaitMembersUp(expected: Set[Address], got: Set[Address] = Set.empty): Unit = {
val members = got + expectMsgType[MemberUp].member.address
if (members != expected) awaitMembersUp(expected, members)
}
awaitMembersUp(Set(firstAddress, secondAddress, thirdAddress))
Cluster(system).unsubscribe(testActor) Cluster(system).unsubscribe(testActor)

10
akka-samples/akka-sample-cluster/src/multi-jvm/scala/sample/cluster/stats/japi/StatsSampleSingleMasterJapiSpec.scala
View file

@ -80,14 +80,8 @@ abstract class StatsSampleSingleMasterJapiSpec extends MultiNodeSpec(StatsSample
Cluster(system) join firstAddress Cluster(system) join firstAddress
// FIXME ticket 3239 duplicate MemberUp events, it should be possible to use receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (
// receiveN(3).collect { case MemberUp(m) => m.address }.toSet must be (Set(firstAddress, secondAddress, thirdAddress)) Set(firstAddress, secondAddress, thirdAddress))
import akka.actor.Address
@scala.annotation.tailrec def awaitMembersUp(expected: Set[Address], got: Set[Address] = Set.empty): Unit = {
val members = got + expectMsgType[MemberUp].member.address
if (members != expected) awaitMembersUp(expected, members)
}
awaitMembersUp(Set(firstAddress, secondAddress, thirdAddress))
Cluster(system).unsubscribe(testActor) Cluster(system).unsubscribe(testActor)