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

Publish LeaderChanged when first seen, see #3131

Browse source 
* The problem in ClusterSingletonManagerChaosSpec was that node 4 doesn't publish
  LeaderChanged, because there is never convergence on node 4 of the new Up
  state for the three new nodes before they are shutdown. When it becomes
  convergence on node 4 prevConvergedGossip and newGossip have same leader
  (i.e. no change).
* LeaderChanged is now published when the new leader is first seen, i.e. same
  as member events. This makes sense now when leader can't be in Joining state.
This commit is contained in:
Patrik Nordwall 2013-03-11 12:41:15 +01:00
parent 369811981e
commit 1e4b2585c7
2 changed files with 9 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

20
akka-cluster/src/main/scala/akka/cluster/ClusterEvent.scala
View file

@ -96,7 +96,8 @@ object ClusterEvent {
}
/**
* Leader of the cluster members changed. Only published after convergence.
* Leader of the cluster members changed. Published when the state change
* is first seen on a node.
*/
case class LeaderChanged(leader: Option[Address]) extends ClusterDomainEvent {
/**
@ -210,8 +211,6 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
import InternalClusterAction._
var latestGossip: Gossip = Gossip.empty
var latestConvergedGossip: Gossip = Gossip.empty
var bufferedEvents: immutable.IndexedSeq[ClusterDomainEvent] = Vector.empty
override def preRestart(reason: Throwable, message: Option[Any]) {
// don't postStop when restarted, no children to stop
@ -243,7 +242,7 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
members = latestGossip.members,
unreachable = latestGossip.overview.unreachable,
seenBy = latestGossip.seenBy,
leader = latestConvergedGossip.leader)
leader = latestGossip.leader)
receiver match {
case Some(ref) ⇒ ref ! state
case None ⇒ publish(state)
@ -275,15 +274,7 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
case _ ⇒ publish(event)
}
}
// buffer up the LeaderChanged waiting for convergence
bufferedEvents ++= diffLeader(oldGossip, newGossip)
// if we have convergence then publish the MemberEvents and LeaderChanged
if (newGossip.convergence) {
val previousConvergedGossip = latestConvergedGossip
latestConvergedGossip = newGossip
bufferedEvents foreach publish
bufferedEvents = Vector.empty
}
diffLeader(oldGossip, newGossip) foreach publish
// publish internal SeenState for testing purposes
diffSeen(oldGossip, newGossip) foreach publish
}
@ -293,13 +284,12 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
def publish(event: AnyRef): Unit = eventStream publish event
def publishStart(): Unit =
if ((latestGossip ne Gossip.empty) || (latestConvergedGossip ne Gossip.empty)) {
if (latestGossip ne Gossip.empty) {
clearState()
publishCurrentClusterState(None)
}
def clearState(): Unit = {
latestGossip = Gossip.empty
latestConvergedGossip = Gossip.empty
}
}