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

Node that joins again should be ignored, see #2184

Browse source
This commit is contained in:
Patrik Nordwall 2012-06-07 15:45:10 +02:00
parent 7e55c8424b
commit dbac17621f
2 changed files with 45 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

31
akka-cluster/src/main/scala/akka/cluster/Cluster.scala
View file

@ -571,27 +571,28 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val localState = state.get
val localGossip = localState.latestGossip
val localMembers = localGossip.members
val localOverview = localGossip.overview
val localUnreachableMembers = localOverview.unreachable
// remove the node from the 'unreachable' set in case it is a DOWN node that is rejoining cluster
val newUnreachableMembers = localUnreachableMembers filterNot { _.address == node }
val newOverview = localOverview copy (unreachable = newUnreachableMembers)
if (!localMembers.exists(_.address == node)) {
val newMembers = localMembers + Member(node, MemberStatus.Joining) // add joining node as Joining
val newGossip = localGossip copy (overview = newOverview, members = newMembers)
// remove the node from the 'unreachable' set in case it is a DOWN node that is rejoining cluster
val newUnreachableMembers = localGossip.overview.unreachable filterNot { _.address == node }
val newOverview = localGossip.overview copy (unreachable = newUnreachableMembers)
val versionedGossip = newGossip + vclockNode
val seenVersionedGossip = versionedGossip seen selfAddress
val newMembers = localMembers + Member(node, MemberStatus.Joining) // add joining node as Joining
val newGossip = localGossip copy (overview = newOverview, members = newMembers)
val newState = localState copy (latestGossip = seenVersionedGossip)
val versionedGossip = newGossip + vclockNode
val seenVersionedGossip = versionedGossip seen selfAddress
if (!state.compareAndSet(localState, newState)) joining(node) // recur if we failed update
else {
if (node != selfAddress) failureDetector heartbeat node
val newState = localState copy (latestGossip = seenVersionedGossip)
if (convergence(newState.latestGossip).isDefined) {
newState.memberMembershipChangeListeners foreach { _ notify newMembers }
if (!state.compareAndSet(localState, newState)) joining(node) // recur if we failed update
else {
if (node != selfAddress) failureDetector heartbeat node
if (convergence(newState.latestGossip).isDefined) {
newState.memberMembershipChangeListeners foreach { _ notify newMembers }
}
}
}
}