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

Refactoring of Gossip class, #23290

Browse source 
* move methods that depends on selfUniqueAddress and selfDc
  to a separate MembershipState class, which also holds the
  latest gossip
* this removes the need to pass in the parameters from everywhere and
  makes it easier to cache some results
* makes it clear that those parameters are always selfUniqueAddress
  and selfDc, instead of some arbitary node/dc
This commit is contained in:
Patrik Nordwall 2017-07-04 21:58:03 +02:00
parent dee14c5b20
commit 867cc97bdd
10 changed files with 386 additions and 321 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-cluster/src/main/scala/akka/cluster/AutoDown.scala
View file

@ -101,7 +101,7 @@ private[cluster] abstract class AutoDownBase(autoDownUnreachableAfter: FiniteDur
import context.dispatcher import context.dispatcher
val skipMemberStatus = Gossip.convergenceSkipUnreachableWithMemberStatus val skipMemberStatus = MembershipState.convergenceSkipUnreachableWithMemberStatus
var scheduledUnreachable: Map[UniqueAddress, Cancellable] = Map.empty var scheduledUnreachable: Map[UniqueAddress, Cancellable] = Map.empty
var pendingUnreachable: Set[UniqueAddress] = Set.empty var pendingUnreachable: Set[UniqueAddress] = Set.empty

96
akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala
View file

@ -13,6 +13,7 @@ import akka.actor._
import akka.actor.SupervisorStrategy.Stop import akka.actor.SupervisorStrategy.Stop
import akka.cluster.MemberStatus._ import akka.cluster.MemberStatus._
import akka.cluster.ClusterEvent._ import akka.cluster.ClusterEvent._
import akka.cluster.ClusterSettings.DataCenter
import akka.dispatch.{ RequiresMessageQueue, UnboundedMessageQueueSemantics } import akka.dispatch.{ RequiresMessageQueue, UnboundedMessageQueueSemantics }
import scala.collection.breakOut import scala.collection.breakOut
@ -157,7 +158,7 @@ private[cluster] object InternalClusterAction {
final case class SendCurrentClusterState(receiver: ActorRef) extends SubscriptionMessage final case class SendCurrentClusterState(receiver: ActorRef) extends SubscriptionMessage
sealed trait PublishMessage sealed trait PublishMessage
final case class PublishChanges(newGossip: Gossip) extends PublishMessage final case class PublishChanges(state: MembershipState) extends PublishMessage
final case class PublishEvent(event: ClusterDomainEvent) extends PublishMessage final case class PublishEvent(event: ClusterDomainEvent) extends PublishMessage
final case object ExitingCompleted final case object ExitingCompleted
@ -277,6 +278,7 @@ private[cluster] object ClusterCoreDaemon {
val NumberOfGossipsBeforeShutdownWhenLeaderExits = 5 val NumberOfGossipsBeforeShutdownWhenLeaderExits = 5
val MaxGossipsBeforeShuttingDownMyself = 5 val MaxGossipsBeforeShuttingDownMyself = 5
} }
/** /**
@ -287,6 +289,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
with RequiresMessageQueue[UnboundedMessageQueueSemantics] { with RequiresMessageQueue[UnboundedMessageQueueSemantics] {
import InternalClusterAction._ import InternalClusterAction._
import ClusterCoreDaemon._ import ClusterCoreDaemon._
import MembershipState._
val cluster = Cluster(context.system) val cluster = Cluster(context.system)
import cluster.{ selfAddress, selfRoles, scheduler, failureDetector } import cluster.{ selfAddress, selfRoles, scheduler, failureDetector }
@ -299,7 +302,8 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
// note that self is not initially member, // note that self is not initially member,
// and the Gossip is not versioned for this 'Node' yet // and the Gossip is not versioned for this 'Node' yet
var latestGossip: Gossip = Gossip.empty var membershipState = MembershipState(Gossip.empty, cluster.selfUniqueAddress, cluster.settings.DataCenter)
def latestGossip: Gossip = membershipState.latestGossip
val statsEnabled = PublishStatsInterval.isFinite val statsEnabled = PublishStatsInterval.isFinite
var gossipStats = GossipStats() var gossipStats = GossipStats()
@ -478,7 +482,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
def initJoin(): Unit = { def initJoin(): Unit = {
val selfStatus = latestGossip.member(selfUniqueAddress).status val selfStatus = latestGossip.member(selfUniqueAddress).status
if (Gossip.removeUnreachableWithMemberStatus.contains(selfStatus)) { if (removeUnreachableWithMemberStatus.contains(selfStatus)) {
// prevents a Down and Exiting node from being used for joining // prevents a Down and Exiting node from being used for joining
logInfo("Sending InitJoinNack message from node [{}] to [{}]", selfAddress, sender()) logInfo("Sending InitJoinNack message from node [{}] to [{}]", selfAddress, sender())
sender() ! InitJoinNack(selfAddress) sender() ! InitJoinNack(selfAddress)
@ -570,7 +574,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
log.warning( log.warning(
"Member with wrong ActorSystem name tried to join, but was ignored, expected [{}] but was [{}]", "Member with wrong ActorSystem name tried to join, but was ignored, expected [{}] but was [{}]",
selfAddress.system, joiningNode.address.system) selfAddress.system, joiningNode.address.system)
else if (Gossip.removeUnreachableWithMemberStatus.contains(selfStatus)) else if (removeUnreachableWithMemberStatus.contains(selfStatus))
logInfo("Trying to join [{}] to [{}] member, ignoring. Use a member that is Up instead.", joiningNode, selfStatus) logInfo("Trying to join [{}] to [{}] member, ignoring. Use a member that is Up instead.", joiningNode, selfStatus)
else { else {
val localMembers = latestGossip.members val localMembers = latestGossip.members
@ -616,7 +620,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
} else } else
sender() ! Welcome(selfUniqueAddress, latestGossip) sender() ! Welcome(selfUniqueAddress, latestGossip)
publish(latestGossip) publishMembershipState()
} }
} }
} }
@ -629,10 +633,10 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
if (joinWith != from.address) if (joinWith != from.address)
logInfo("Ignoring welcome from [{}] when trying to join with [{}]", from.address, joinWith) logInfo("Ignoring welcome from [{}] when trying to join with [{}]", from.address, joinWith)
else { else {
latestGossip = gossip seen selfUniqueAddress membershipState = membershipState.copy(latestGossip = gossip).seen()
logInfo("Welcome from [{}]", from.address) logInfo("Welcome from [{}]", from.address)
assertLatestGossip() assertLatestGossip()
publish(latestGossip) publishMembershipState()
if (from != selfUniqueAddress) if (from != selfUniqueAddress)
gossipTo(from, sender()) gossipTo(from, sender())
becomeInitialized() becomeInitialized()
@ -653,7 +657,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
updateLatestGossip(newGossip) updateLatestGossip(newGossip)
logInfo("Marked address [{}] as [{}]", address, Leaving) logInfo("Marked address [{}] as [{}]", address, Leaving)
publish(latestGossip) publishMembershipState()
// immediate gossip to speed up the leaving process // immediate gossip to speed up the leaving process
gossip() gossip()
} }
@ -664,9 +668,9 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
// ExitingCompleted sent via CoordinatedShutdown to continue the leaving process. // ExitingCompleted sent via CoordinatedShutdown to continue the leaving process.
exitingTasksInProgress = false exitingTasksInProgress = false
// mark as seen // mark as seen
latestGossip = latestGossip seen selfUniqueAddress membershipState = membershipState.seen()
assertLatestGossip() assertLatestGossip()
publish(latestGossip) publishMembershipState()
// Let others know (best effort) before shutdown. Otherwise they will not see // Let others know (best effort) before shutdown. Otherwise they will not see
// convergence of the Exiting state until they have detected this node as // convergence of the Exiting state until they have detected this node as
@ -681,10 +685,10 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
// send ExitingConfirmed to two potential leaders // send ExitingConfirmed to two potential leaders
val membersExceptSelf = latestGossip.members.filter(_.uniqueAddress != selfUniqueAddress) val membersExceptSelf = latestGossip.members.filter(_.uniqueAddress != selfUniqueAddress)
latestGossip.leaderOf(selfDc, membersExceptSelf, selfUniqueAddress) match { membershipState.leaderOf(membersExceptSelf) match {
case Some(node1) ⇒ case Some(node1) ⇒
clusterCore(node1.address) ! ExitingConfirmed(selfUniqueAddress) clusterCore(node1.address) ! ExitingConfirmed(selfUniqueAddress)
latestGossip.leaderOf(selfDc, membersExceptSelf.filterNot(_.uniqueAddress == node1), selfUniqueAddress) match { membershipState.leaderOf(membersExceptSelf.filterNot(_.uniqueAddress == node1)) match {
case Some(node2) ⇒ case Some(node2) ⇒
clusterCore(node2.address) ! ExitingConfirmed(selfUniqueAddress) clusterCore(node2.address) ! ExitingConfirmed(selfUniqueAddress)
case None ⇒ // no more potential leader case None ⇒ // no more potential leader
@ -723,7 +727,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
val localMembers = localGossip.members val localMembers = localGossip.members
val localOverview = localGossip.overview val localOverview = localGossip.overview
val localSeen = localOverview.seen val localSeen = localOverview.seen
val localReachability = localGossip.dcReachability(selfDc) val localReachability = membershipState.dcReachability
// check if the node to DOWN is in the `members` set // check if the node to DOWN is in the `members` set
localMembers.find(_.address == address) match { localMembers.find(_.address == address) match {
@ -735,7 +739,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
val newGossip = localGossip.markAsDown(m) val newGossip = localGossip.markAsDown(m)
updateLatestGossip(newGossip) updateLatestGossip(newGossip)
publish(latestGossip) publishMembershipState()
case Some(_) ⇒ // already down case Some(_) ⇒ // already down
case None ⇒ case None ⇒
logInfo("Ignoring down of unknown node [{}]", address) logInfo("Ignoring down of unknown node [{}]", address)
@ -753,7 +757,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
log.warning( log.warning(
"Cluster Node [{}] - Marking node as TERMINATED [{}], due to quarantine. Node roles [{}]", "Cluster Node [{}] - Marking node as TERMINATED [{}], due to quarantine. Node roles [{}]",
selfAddress, node.address, selfRoles.mkString(",")) selfAddress, node.address, selfRoles.mkString(","))
publish(latestGossip) publishMembershipState()
downing(node.address) downing(node.address)
} }
} }
@ -829,14 +833,14 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
// Perform the same pruning (clear of VectorClock) as the leader did when removing a member. // Perform the same pruning (clear of VectorClock) as the leader did when removing a member.
// Removal of member itself is handled in merge (pickHighestPriority) // Removal of member itself is handled in merge (pickHighestPriority)
val prunedLocalGossip = localGossip.members.foldLeft(localGossip) { (g, m) ⇒ val prunedLocalGossip = localGossip.members.foldLeft(localGossip) { (g, m) ⇒
if (Gossip.removeUnreachableWithMemberStatus(m.status) && !remoteGossip.members.contains(m)) { if (removeUnreachableWithMemberStatus(m.status) && !remoteGossip.members.contains(m)) {
log.debug("Cluster Node [{}] - Pruned conflicting local gossip: {}", selfAddress, m) log.debug("Cluster Node [{}] - Pruned conflicting local gossip: {}", selfAddress, m)
g.prune(VectorClock.Node(vclockName(m.uniqueAddress))) g.prune(VectorClock.Node(vclockName(m.uniqueAddress)))
} else } else
g g
} }
val prunedRemoteGossip = remoteGossip.members.foldLeft(remoteGossip) { (g, m) ⇒ val prunedRemoteGossip = remoteGossip.members.foldLeft(remoteGossip) { (g, m) ⇒
if (Gossip.removeUnreachableWithMemberStatus(m.status) && !localGossip.members.contains(m)) { if (removeUnreachableWithMemberStatus(m.status) && !localGossip.members.contains(m)) {
log.debug("Cluster Node [{}] - Pruned conflicting remote gossip: {}", selfAddress, m) log.debug("Cluster Node [{}] - Pruned conflicting remote gossip: {}", selfAddress, m)
g.prune(VectorClock.Node(vclockName(m.uniqueAddress))) g.prune(VectorClock.Node(vclockName(m.uniqueAddress)))
} else } else
@ -849,9 +853,9 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
// Don't mark gossip state as seen while exiting is in progress, e.g. // Don't mark gossip state as seen while exiting is in progress, e.g.
// shutting down singleton actors. This delays removal of the member until // shutting down singleton actors. This delays removal of the member until
// the exiting tasks have been completed. // the exiting tasks have been completed.
latestGossip = membershipState = membershipState.copy(latestGossip =
if (exitingTasksInProgress) winningGossip if (exitingTasksInProgress) winningGossip
else winningGossip seen selfUniqueAddress else winningGossip seen selfUniqueAddress)
assertLatestGossip() assertLatestGossip()
// for all new joining nodes we remove them from the failure detector // for all new joining nodes we remove them from the failure detector
@ -877,7 +881,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
} }
} }
publish(latestGossip) publishMembershipState()
val selfStatus = latestGossip.member(selfUniqueAddress).status val selfStatus = latestGossip.member(selfUniqueAddress).status
if (selfStatus == Exiting && !exitingTasksInProgress) { if (selfStatus == Exiting && !exitingTasksInProgress) {
@ -1004,11 +1008,11 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
* Runs periodic leader actions, such as member status transitions, assigning partitions etc. * Runs periodic leader actions, such as member status transitions, assigning partitions etc.
*/ */
def leaderActions(): Unit = { def leaderActions(): Unit = {
if (latestGossip.isDcLeader(selfDc, selfUniqueAddress, selfUniqueAddress)) { if (membershipState.isLeader(selfUniqueAddress)) {
// only run the leader actions if we are the LEADER of the data center // only run the leader actions if we are the LEADER of the data center
val firstNotice = 20 val firstNotice = 20
val periodicNotice = 60 val periodicNotice = 60
if (latestGossip.convergence(selfDc, selfUniqueAddress, exitingConfirmed)) { if (membershipState.convergence(exitingConfirmed)) {
if (leaderActionCounter >= firstNotice) if (leaderActionCounter >= firstNotice)
logInfo("Leader can perform its duties again") logInfo("Leader can perform its duties again")
leaderActionCounter = 0 leaderActionCounter = 0
@ -1021,7 +1025,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
if (leaderActionCounter == firstNotice || leaderActionCounter % periodicNotice == 0) if (leaderActionCounter == firstNotice || leaderActionCounter % periodicNotice == 0)
logInfo( logInfo(
"Leader can currently not perform its duties, reachability status: [{}], member status: [{}]", "Leader can currently not perform its duties, reachability status: [{}], member status: [{}]",
latestGossip.dcReachabilityExcludingDownedObservers(selfDc), membershipState.dcReachabilityExcludingDownedObservers,
latestGossip.members.collect { latestGossip.members.collect {
case m if m.dataCenter == selfDc ⇒ case m if m.dataCenter == selfDc ⇒
s"${m.address} ${m.status} seen=${latestGossip.seenByNode(m.uniqueAddress)}" s"${m.address} ${m.status} seen=${latestGossip.seenByNode(m.uniqueAddress)}"
@ -1036,8 +1040,8 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
if (latestGossip.member(selfUniqueAddress).status == Down) { if (latestGossip.member(selfUniqueAddress).status == Down) {
// When all reachable have seen the state this member will shutdown itself when it has // When all reachable have seen the state this member will shutdown itself when it has
// status Down. The down commands should spread before we shutdown. // status Down. The down commands should spread before we shutdown.
val unreachable = latestGossip.dcReachability(selfDc).allUnreachableOrTerminated val unreachable = membershipState.dcReachability.allUnreachableOrTerminated
val downed = latestGossip.dcMembers(selfDc).collect { case m if m.status == Down ⇒ m.uniqueAddress } val downed = membershipState.dcMembers.collect { case m if m.status == Down ⇒ m.uniqueAddress }
if (downed.forall(node ⇒ unreachable(node) || latestGossip.seenByNode(node))) { if (downed.forall(node ⇒ unreachable(node) || latestGossip.seenByNode(node))) {
// the reason for not shutting down immediately is to give the gossip a chance to spread // the reason for not shutting down immediately is to give the gossip a chance to spread
// the downing information to other downed nodes, so that they can shutdown themselves // the downing information to other downed nodes, so that they can shutdown themselves
@ -1072,9 +1076,9 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
def leaderActionsOnConvergence(): Unit = { def leaderActionsOnConvergence(): Unit = {
val removedUnreachable = for { val removedUnreachable = for {
node ← latestGossip.dcReachability(selfDc).allUnreachableOrTerminated node ← membershipState.dcReachability.allUnreachableOrTerminated
m = latestGossip.member(node) m = latestGossip.member(node)
if m.dataCenter == selfDc && Gossip.removeUnreachableWithMemberStatus(m.status) if m.dataCenter == selfDc && removeUnreachableWithMemberStatus(m.status)
} yield m } yield m
val removedExitingConfirmed = exitingConfirmed.filter { n ⇒ val removedExitingConfirmed = exitingConfirmed.filter { n ⇒
@ -1148,7 +1152,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
val pruned = updatedGossip.pruneTombstones(System.currentTimeMillis() - PruneGossipTombstonesAfter.toMillis) val pruned = updatedGossip.pruneTombstones(System.currentTimeMillis() - PruneGossipTombstonesAfter.toMillis)
if (pruned ne latestGossip) { if (pruned ne latestGossip) {
updateLatestGossip(pruned) updateLatestGossip(pruned)
publish(pruned) publishMembershipState()
} }
} }
@ -1161,7 +1165,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
m.dataCenter == selfDc && m.dataCenter == selfDc &&
m.status == Joining && m.status == Joining &&
enoughMembers && enoughMembers &&
latestGossip.dcReachabilityExcludingDownedObservers(selfDc).isReachable(m.uniqueAddress) membershipState.dcReachabilityExcludingDownedObservers.isReachable(m.uniqueAddress)
val changedMembers = localMembers.collect { val changedMembers = localMembers.collect {
case m if isJoiningToWeaklyUp(m) ⇒ m.copy(status = WeaklyUp) case m if isJoiningToWeaklyUp(m) ⇒ m.copy(status = WeaklyUp)
} }
@ -1177,7 +1181,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
logInfo("Leader is moving node [{}] to [{}]", m.address, m.status) logInfo("Leader is moving node [{}] to [{}]", m.address, m.status)
} }
publish(latestGossip) publishMembershipState()
} }
} }
@ -1230,7 +1234,7 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
if (newlyDetectedReachableMembers.nonEmpty) if (newlyDetectedReachableMembers.nonEmpty)
logInfo("Marking node(s) as REACHABLE [{}]. Node roles [{}]", newlyDetectedReachableMembers.mkString(", "), selfRoles.mkString(",")) logInfo("Marking node(s) as REACHABLE [{}]. Node roles [{}]", newlyDetectedReachableMembers.mkString(", "), selfRoles.mkString(","))
publish(latestGossip) publishMembershipState()
} }
} }
} }
@ -1269,23 +1273,25 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
clusterCore(node.address) ! GossipStatus(selfUniqueAddress, latestGossip.version) clusterCore(node.address) ! GossipStatus(selfUniqueAddress, latestGossip.version)
def validNodeForGossip(node: UniqueAddress): Boolean = def validNodeForGossip(node: UniqueAddress): Boolean =
node != selfUniqueAddress && latestGossip.isReachableExcludingDownedObservers(selfDc, node) node != selfUniqueAddress && membershipState.isReachableExcludingDownedObservers(node)
def updateLatestGossip(newGossip: Gossip): Unit = { def updateLatestGossip(gossip: Gossip): Unit = {
// Updating the vclock version for the changes // Updating the vclock version for the changes
val versionedGossip = newGossip :+ vclockNode val versionedGossip = gossip :+ vclockNode
// Don't mark gossip state as seen while exiting is in progress, e.g. // Don't mark gossip state as seen while exiting is in progress, e.g.
// shutting down singleton actors. This delays removal of the member until // shutting down singleton actors. This delays removal of the member until
// the exiting tasks have been completed. // the exiting tasks have been completed.
if (exitingTasksInProgress) val newGossip =
latestGossip = versionedGossip.clearSeen() if (exitingTasksInProgress)
else { versionedGossip.clearSeen()
// Nobody else has seen this gossip but us else {
val seenVersionedGossip = versionedGossip onlySeen (selfUniqueAddress) // Nobody else has seen this gossip but us
// Update the state with the new gossip val seenVersionedGossip = versionedGossip onlySeen (selfUniqueAddress)
latestGossip = seenVersionedGossip // Update the state with the new gossip
} seenVersionedGossip
}
membershipState = membershipState.copy(newGossip)
assertLatestGossip() assertLatestGossip()
} }
@ -1293,11 +1299,11 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef) extends Actor with
if (Cluster.isAssertInvariantsEnabled && latestGossip.version.versions.size > latestGossip.members.size) if (Cluster.isAssertInvariantsEnabled && latestGossip.version.versions.size > latestGossip.members.size)
throw new IllegalStateException(s"Too many vector clock entries in gossip state ${latestGossip}") throw new IllegalStateException(s"Too many vector clock entries in gossip state ${latestGossip}")
def publish(newGossip: Gossip): Unit = { def publishMembershipState(): Unit = {
if (cluster.settings.Debug.VerboseGossipLogging) if (cluster.settings.Debug.VerboseGossipLogging)
log.debug("Cluster Node [{}] dc [{}] - New gossip published [{}]", selfAddress, cluster.settings.DataCenter, newGossip) log.debug("Cluster Node [{}] dc [{}] - New gossip published [{}]", selfAddress, cluster.settings.DataCenter, membershipState.latestGossip)
publisher ! PublishChanges(newGossip) publisher ! PublishChanges(membershipState)
if (PublishStatsInterval == Duration.Zero) publishInternalStats() if (PublishStatsInterval == Duration.Zero) publishInternalStats()
} }

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

@ -265,12 +265,14 @@ object ClusterEvent {
/** /**
* INTERNAL API * INTERNAL API
*/ */
private[cluster] def diffUnreachable(oldGossip: Gossip, newGossip: Gossip, selfUniqueAddress: UniqueAddress): immutable.Seq[UnreachableMember] = private[cluster] def diffUnreachable(oldState: MembershipState, newState: MembershipState): immutable.Seq[UnreachableMember] =
if (newGossip eq oldGossip) Nil if (newState eq oldState) Nil
else { else {
val oldGossip = oldState.latestGossip
val newGossip = newState.latestGossip
val oldUnreachableNodes = oldGossip.overview.reachability.allUnreachableOrTerminated val oldUnreachableNodes = oldGossip.overview.reachability.allUnreachableOrTerminated
(newGossip.overview.reachability.allUnreachableOrTerminated.collect { (newGossip.overview.reachability.allUnreachableOrTerminated.collect {
case node if !oldUnreachableNodes.contains(node) && node != selfUniqueAddress ⇒ case node if !oldUnreachableNodes.contains(node) && node != newState.selfUniqueAddress ⇒
UnreachableMember(newGossip.member(node)) UnreachableMember(newGossip.member(node))
})(collection.breakOut) })(collection.breakOut)
} }
@ -278,11 +280,13 @@ object ClusterEvent {
/** /**
* INTERNAL API * INTERNAL API
*/ */
private[cluster] def diffReachable(oldGossip: Gossip, newGossip: Gossip, selfUniqueAddress: UniqueAddress): immutable.Seq[ReachableMember] = private[cluster] def diffReachable(oldState: MembershipState, newState: MembershipState): immutable.Seq[ReachableMember] =
if (newGossip eq oldGossip) Nil if (newState eq oldState) Nil
else { else {
(oldGossip.overview.reachability.allUnreachable.collect { val oldGossip = oldState.latestGossip
case node if newGossip.hasMember(node) && newGossip.overview.reachability.isReachable(node) && node != selfUniqueAddress ⇒ val newGossip = newState.latestGossip
(oldState.overview.reachability.allUnreachable.collect {
case node if newGossip.hasMember(node) && newGossip.overview.reachability.isReachable(node) && node != newState.selfUniqueAddress ⇒
ReachableMember(newGossip.member(node)) ReachableMember(newGossip.member(node))
})(collection.breakOut) })(collection.breakOut)
@ -291,9 +295,11 @@ object ClusterEvent {
/** /**
* INTERNAL API. * INTERNAL API.
*/ */
private[cluster] def diffMemberEvents(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[MemberEvent] = private[cluster] def diffMemberEvents(oldState: MembershipState, newState: MembershipState): immutable.Seq[MemberEvent] =
if (newGossip eq oldGossip) Nil if (newState eq oldState) Nil
else { else {
val oldGossip = oldState.latestGossip
val newGossip = newState.latestGossip
val newMembers = newGossip.members diff oldGossip.members val newMembers = newGossip.members diff oldGossip.members
val membersGroupedByAddress = List(newGossip.members, oldGossip.members).flatten.groupBy(_.uniqueAddress) val membersGroupedByAddress = List(newGossip.members, oldGossip.members).flatten.groupBy(_.uniqueAddress)
val changedMembers = membersGroupedByAddress collect { val changedMembers = membersGroupedByAddress collect {
@ -319,9 +325,9 @@ object ClusterEvent {
* INTERNAL API * INTERNAL API
*/ */
@InternalApi @InternalApi
private[cluster] def diffLeader(dc: DataCenter, oldGossip: Gossip, newGossip: Gossip, selfUniqueAddress: UniqueAddress): immutable.Seq[LeaderChanged] = { private[cluster] def diffLeader(oldState: MembershipState, newState: MembershipState): immutable.Seq[LeaderChanged] = {
val newLeader = newGossip.dcLeader(dc, selfUniqueAddress) val newLeader = newState.leader
if (newLeader != oldGossip.dcLeader(dc, selfUniqueAddress)) List(LeaderChanged(newLeader.map(_.address))) if (newLeader != oldState.leader) List(LeaderChanged(newLeader.map(_.address)))
else Nil else Nil
} }
@ -329,11 +335,11 @@ object ClusterEvent {
* INTERNAL API * INTERNAL API
*/ */
@InternalApi @InternalApi
private[cluster] def diffRolesLeader(dc: DataCenter, oldGossip: Gossip, newGossip: Gossip, selfUniqueAddress: UniqueAddress): Set[RoleLeaderChanged] = { private[cluster] def diffRolesLeader(oldState: MembershipState, newState: MembershipState): Set[RoleLeaderChanged] = {
for { for {
role ← oldGossip.allRoles union newGossip.allRoles role ← oldState.latestGossip.allRoles union newState.latestGossip.allRoles
newLeader = newGossip.roleLeader(dc, role, selfUniqueAddress) newLeader = newState.roleLeader(role)
if newLeader != oldGossip.roleLeader(dc, role, selfUniqueAddress) if newLeader != oldState.roleLeader(role)
} yield RoleLeaderChanged(role, newLeader.map(_.address)) } yield RoleLeaderChanged(role, newLeader.map(_.address))
} }
@ -341,12 +347,12 @@ object ClusterEvent {
* INTERNAL API * INTERNAL API
*/ */
@InternalApi @InternalApi
private[cluster] def diffSeen(dc: DataCenter, oldGossip: Gossip, newGossip: Gossip, selfUniqueAddress: UniqueAddress): immutable.Seq[SeenChanged] = private[cluster] def diffSeen(oldState: MembershipState, newState: MembershipState): immutable.Seq[SeenChanged] =
if (newGossip eq oldGossip) Nil if (oldState eq newState) Nil
else { else {
val newConvergence = newGossip.convergence(dc, selfUniqueAddress, Set.empty) val newConvergence = newState.convergence(Set.empty)
val newSeenBy = newGossip.seenBy val newSeenBy = newState.latestGossip.seenBy
if (newConvergence != oldGossip.convergence(dc, selfUniqueAddress, Set.empty) || newSeenBy != oldGossip.seenBy) if (newConvergence != oldState.convergence(Set.empty) || newSeenBy != oldState.latestGossip.seenBy)
List(SeenChanged(newConvergence, newSeenBy.map(_.address))) List(SeenChanged(newConvergence, newSeenBy.map(_.address)))
else Nil else Nil
} }
@ -355,9 +361,9 @@ object ClusterEvent {
* INTERNAL API * INTERNAL API
*/ */
@InternalApi @InternalApi
private[cluster] def diffReachability(oldGossip: Gossip, newGossip: Gossip): immutable.Seq[ReachabilityChanged] = private[cluster] def diffReachability(oldState: MembershipState, newState: MembershipState): immutable.Seq[ReachabilityChanged] =
if (newGossip.overview.reachability eq oldGossip.overview.reachability) Nil if (newState.overview.reachability eq oldState.overview.reachability) Nil
else List(ReachabilityChanged(newGossip.overview.reachability)) else List(ReachabilityChanged(newState.overview.reachability))
} }
@ -372,7 +378,8 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
val cluster = Cluster(context.system) val cluster = Cluster(context.system)
val selfUniqueAddress = cluster.selfUniqueAddress val selfUniqueAddress = cluster.selfUniqueAddress
var latestGossip: Gossip = Gossip.empty val emptyMembershipState = MembershipState(Gossip.empty, cluster.selfUniqueAddress, cluster.settings.DataCenter)
var membershipState: MembershipState = emptyMembershipState
def selfDc = cluster.settings.DataCenter def selfDc = cluster.settings.DataCenter
override def preRestart(reason: Throwable, message: Option[Any]) { override def preRestart(reason: Throwable, message: Option[Any]) {
@ -382,11 +389,11 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
override def postStop(): Unit = { override def postStop(): Unit = {
// publish the final removed state before shutting down // publish the final removed state before shutting down
publish(ClusterShuttingDown) publish(ClusterShuttingDown)
publishChanges(Gossip.empty) publishChanges(emptyMembershipState)
} }
def receive = { def receive = {
case PublishChanges(newGossip) ⇒ publishChanges(newGossip) case PublishChanges(newState) ⇒ publishChanges(newState)
case currentStats: CurrentInternalStats ⇒ publishInternalStats(currentStats) case currentStats: CurrentInternalStats ⇒ publishInternalStats(currentStats)
case SendCurrentClusterState(receiver) ⇒ sendCurrentClusterState(receiver) case SendCurrentClusterState(receiver) ⇒ sendCurrentClusterState(receiver)
case Subscribe(subscriber, initMode, to) ⇒ subscribe(subscriber, initMode, to) case Subscribe(subscriber, initMode, to) ⇒ subscribe(subscriber, initMode, to)
@ -401,16 +408,17 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
* to mimic what you would have seen if you were listening to the events. * to mimic what you would have seen if you were listening to the events.
*/ */
def sendCurrentClusterState(receiver: ActorRef): Unit = { def sendCurrentClusterState(receiver: ActorRef): Unit = {
val unreachable: Set[Member] = latestGossip.overview.reachability.allUnreachableOrTerminated.collect { val unreachable: Set[Member] =
case node if node != selfUniqueAddress ⇒ latestGossip.member(node) membershipState.latestGossip.overview.reachability.allUnreachableOrTerminated.collect {
} case node if node != selfUniqueAddress ⇒ membershipState.latestGossip.member(node)
}
val state = CurrentClusterState( val state = CurrentClusterState(
members = latestGossip.members, members = membershipState.latestGossip.members,
unreachable = unreachable, unreachable = unreachable,
seenBy = latestGossip.seenBy.map(_.address), seenBy = membershipState.latestGossip.seenBy.map(_.address),
leader = latestGossip.dcLeader(selfDc, selfUniqueAddress).map(_.address), leader = membershipState.leader.map(_.address),
roleLeaderMap = latestGossip.allRoles.map(r ⇒ roleLeaderMap = membershipState.latestGossip.allRoles.map(r ⇒
r → latestGossip.roleLeader(selfDc, r, selfUniqueAddress).map(_.address))(collection.breakOut)) r → membershipState.roleLeader(r).map(_.address))(collection.breakOut))
receiver ! state receiver ! state
} }
@ -421,7 +429,7 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
if (to.exists(_.isAssignableFrom(event.getClass))) if (to.exists(_.isAssignableFrom(event.getClass)))
subscriber ! event subscriber ! event
} }
publishDiff(Gossip.empty, latestGossip, pub) publishDiff(emptyMembershipState, membershipState, pub)
case InitialStateAsSnapshot ⇒ case InitialStateAsSnapshot ⇒
sendCurrentClusterState(subscriber) sendCurrentClusterState(subscriber)
} }
@ -434,22 +442,22 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
case Some(c) ⇒ eventStream.unsubscribe(subscriber, c) case Some(c) ⇒ eventStream.unsubscribe(subscriber, c)
} }
def publishChanges(newGossip: Gossip): Unit = { def publishChanges(newState: MembershipState): Unit = {
val oldGossip = latestGossip val oldState = membershipState
// keep the latestGossip to be sent to new subscribers // keep the latest state to be sent to new subscribers
latestGossip = newGossip membershipState = newState
publishDiff(oldGossip, newGossip, publish) publishDiff(oldState, newState, publish)
} }
def publishDiff(oldGossip: Gossip, newGossip: Gossip, pub: AnyRef ⇒ Unit): Unit = { def publishDiff(oldState: MembershipState, newState: MembershipState, pub: AnyRef ⇒ Unit): Unit = {
diffMemberEvents(oldGossip, newGossip) foreach pub diffMemberEvents(oldState, newState) foreach pub
diffUnreachable(oldGossip, newGossip, selfUniqueAddress) foreach pub diffUnreachable(oldState, newState) foreach pub
diffReachable(oldGossip, newGossip, selfUniqueAddress) foreach pub diffReachable(oldState, newState) foreach pub
diffLeader(selfDc, oldGossip, newGossip, selfUniqueAddress) foreach pub diffLeader(oldState, newState) foreach pub
diffRolesLeader(selfDc, oldGossip, newGossip, selfUniqueAddress) foreach pub diffRolesLeader(oldState, newState) foreach pub
// publish internal SeenState for testing purposes // publish internal SeenState for testing purposes
diffSeen(selfDc, oldGossip, newGossip, selfUniqueAddress) foreach pub diffSeen(oldState, newState) foreach pub
diffReachability(oldGossip, newGossip) foreach pub diffReachability(oldState, newState) foreach pub
} }
def publishInternalStats(currentStats: CurrentInternalStats): Unit = publish(currentStats) def publishInternalStats(currentStats: CurrentInternalStats): Unit = publish(currentStats)
@ -457,6 +465,6 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
def publish(event: AnyRef): Unit = eventStream publish event def publish(event: AnyRef): Unit = eventStream publish event
def clearState(): Unit = { def clearState(): Unit = {
latestGossip = Gossip.empty membershipState = emptyMembershipState
} }
} }

93
akka-cluster/src/main/scala/akka/cluster/Gossip.scala
View file

@ -22,11 +22,6 @@ private[cluster] object Gossip {
def apply(members: immutable.SortedSet[Member]) = def apply(members: immutable.SortedSet[Member]) =
if (members.isEmpty) empty else empty.copy(members = members) if (members.isEmpty) empty else empty.copy(members = members)
private val leaderMemberStatus = Set[MemberStatus](Up, Leaving)
private val convergenceMemberStatus = Set[MemberStatus](Up, Leaving)
val convergenceSkipUnreachableWithMemberStatus = Set[MemberStatus](Down, Exiting)
val removeUnreachableWithMemberStatus = Set[MemberStatus](Down, Exiting)
} }
/** /**
@ -75,7 +70,7 @@ private[cluster] final case class Gossip(
private def assertInvariants(): Unit = { private def assertInvariants(): Unit = {
if (members.exists(_.status == Removed)) if (members.exists(_.status == Removed))
throw new IllegalArgumentException(s"Live members must have status [${Removed}], " + throw new IllegalArgumentException(s"Live members must not have status [${Removed}], " +
s"got [${members.filter(_.status == Removed)}]") s"got [${members.filter(_.status == Removed)}]")
val inReachabilityButNotMember = overview.reachability.allObservers diff members.map(_.uniqueAddress) val inReachabilityButNotMember = overview.reachability.allObservers diff members.map(_.uniqueAddress)
@ -168,103 +163,17 @@ private[cluster] final case class Gossip(
Gossip(mergedMembers, GossipOverview(mergedSeen, mergedReachability), mergedVClock, mergedTombstones) Gossip(mergedMembers, GossipOverview(mergedSeen, mergedReachability), mergedVClock, mergedTombstones)
} }
/**
* Checks if we have a cluster convergence. If there are any in data center node pairs that cannot reach each other
* then we can't have a convergence until those nodes reach each other again or one of them is downed
*
* @return true if convergence have been reached and false if not
*/
def convergence(dc: DataCenter, selfUniqueAddress: UniqueAddress, exitingConfirmed: Set[UniqueAddress]): Boolean = {
// Find cluster members in the data center that are unreachable from other members of the data center
// excluding observations from members outside of the data center, that have status DOWN or is passed in as confirmed exiting.
val unreachableInDc = dcReachabilityExcludingDownedObservers(dc).allUnreachableOrTerminated.collect {
case node if node != selfUniqueAddress && !exitingConfirmed(node) ⇒ member(node)
}
// If another member in the data center that is UP or LEAVING and has not seen this gossip or is exiting
// convergence cannot be reached
def memberHinderingConvergenceExists =
members.exists(member ⇒
member.dataCenter == dc &&
Gossip.convergenceMemberStatus(member.status) &&
!(seenByNode(member.uniqueAddress) || exitingConfirmed(member.uniqueAddress)))
// unreachables outside of the data center or with status DOWN or EXITING does not affect convergence
def allUnreachablesCanBeIgnored =
unreachableInDc.forall(unreachable ⇒ Gossip.convergenceSkipUnreachableWithMemberStatus(unreachable.status))
allUnreachablesCanBeIgnored && !memberHinderingConvergenceExists
}
lazy val reachabilityExcludingDownedObservers: Reachability = { lazy val reachabilityExcludingDownedObservers: Reachability = {
val downed = members.collect { case m if m.status == Down ⇒ m } val downed = members.collect { case m if m.status == Down ⇒ m }
overview.reachability.removeObservers(downed.map(_.uniqueAddress)) overview.reachability.removeObservers(downed.map(_.uniqueAddress))
} }
/**
* @return Reachability excluding observations from nodes outside of the data center, but including observed unreachable
* nodes outside of the data center
*/
def dcReachability(dc: DataCenter): Reachability =
overview.reachability.removeObservers(members.collect { case m if m.dataCenter != dc ⇒ m.uniqueAddress })
/**
* @return reachability for data center nodes, with observations from outside the data center or from downed nodes filtered out
*/
def dcReachabilityExcludingDownedObservers(dc: DataCenter): Reachability = {
val membersToExclude = members.collect { case m if m.status == Down || m.dataCenter != dc ⇒ m.uniqueAddress }
overview.reachability.removeObservers(membersToExclude).remove(members.collect { case m if m.dataCenter != dc ⇒ m.uniqueAddress })
}
def dcMembers(dc: DataCenter): SortedSet[Member] =
members.filter(_.dataCenter == dc)
def isDcLeader(dc: DataCenter, node: UniqueAddress, selfUniqueAddress: UniqueAddress): Boolean =
dcLeader(dc, selfUniqueAddress).contains(node)
def dcLeader(dc: DataCenter, selfUniqueAddress: UniqueAddress): Option[UniqueAddress] =
leaderOf(dc, members, selfUniqueAddress)
def roleLeader(dc: DataCenter, role: String, selfUniqueAddress: UniqueAddress): Option[UniqueAddress] =
leaderOf(dc, members.filter(_.hasRole(role)), selfUniqueAddress)
def leaderOf(dc: DataCenter, mbrs: immutable.SortedSet[Member], selfUniqueAddress: UniqueAddress): Option[UniqueAddress] = {
val reachability = dcReachability(dc)
val reachableMembersInDc =
if (reachability.isAllReachable) mbrs.filter(m ⇒ m.dataCenter == dc && m.status != Down)
else mbrs.filter(m ⇒
m.dataCenter == dc &&
m.status != Down &&
(reachability.isReachable(m.uniqueAddress) || m.uniqueAddress == selfUniqueAddress))
if (reachableMembersInDc.isEmpty) None
else reachableMembersInDc.find(m ⇒ Gossip.leaderMemberStatus(m.status))
.orElse(Some(reachableMembersInDc.min(Member.leaderStatusOrdering)))
.map(_.uniqueAddress)
}
def allDataCenters: Set[DataCenter] = members.map(_.dataCenter) def allDataCenters: Set[DataCenter] = members.map(_.dataCenter)
def allRoles: Set[String] = members.flatMap(_.roles) def allRoles: Set[String] = members.flatMap(_.roles)
def isSingletonCluster: Boolean = members.size == 1 def isSingletonCluster: Boolean = members.size == 1
/**
* @return true if toAddress should be reachable from the fromDc in general, within a data center
* this means only caring about data center local observations, across data centers it
* means caring about all observations for the toAddress.
*/
def isReachableExcludingDownedObservers(fromDc: DataCenter, toAddress: UniqueAddress): Boolean =
if (!hasMember(toAddress)) false
else {
val to = member(toAddress)
// if member is in the same data center, we ignore cross data center unreachability
if (fromDc == to.dataCenter) dcReachabilityExcludingDownedObservers(fromDc).isReachable(toAddress)
// if not it is enough that any non-downed node observed it as unreachable
else reachabilityExcludingDownedObservers.isReachable(toAddress)
}
/** /**
* @return true if fromAddress should be able to reach toAddress based on the unreachability data and their * @return true if fromAddress should be able to reach toAddress based on the unreachability data and their
* respective data centers * respective data centers

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

@ -163,7 +163,7 @@ object Member {
if (members.size == 2) acc + members.reduceLeft(highestPriorityOf) if (members.size == 2) acc + members.reduceLeft(highestPriorityOf)
else { else {
val m = members.head val m = members.head
if (tombstones.contains(m.uniqueAddress) || Gossip.removeUnreachableWithMemberStatus(m.status)) acc // removed if (tombstones.contains(m.uniqueAddress) || MembershipState.removeUnreachableWithMemberStatus(m.status)) acc // removed
else acc + m else acc + m
} }
} }

122
akka-cluster/src/main/scala/akka/cluster/MembershipState.scala Normal file
View file

@ -0,0 +1,122 @@
/**
* Copyright (C) 2017 Lightbend Inc. <http://www.lightbend.com>
*/
package akka.cluster
import scala.collection.immutable
import scala.collection.SortedSet
import akka.cluster.ClusterSettings.DataCenter
import akka.cluster.MemberStatus._
import akka.annotation.InternalApi
/**
* INTERNAL API
*/
@InternalApi private[akka] object MembershipState {
import MemberStatus._
private val leaderMemberStatus = Set[MemberStatus](Up, Leaving)
private val convergenceMemberStatus = Set[MemberStatus](Up, Leaving)
val convergenceSkipUnreachableWithMemberStatus = Set[MemberStatus](Down, Exiting)
val removeUnreachableWithMemberStatus = Set[MemberStatus](Down, Exiting)
}
/**
* INTERNAL API
*/
@InternalApi private[akka] final case class MembershipState(latestGossip: Gossip, selfUniqueAddress: UniqueAddress, selfDc: DataCenter) {
import MembershipState._
def members: immutable.SortedSet[Member] = latestGossip.members
def overview: GossipOverview = latestGossip.overview
def seen(): MembershipState = copy(latestGossip = latestGossip.seen(selfUniqueAddress))
/**
* Checks if we have a cluster convergence. If there are any in data center node pairs that cannot reach each other
* then we can't have a convergence until those nodes reach each other again or one of them is downed
*
* @return true if convergence have been reached and false if not
*/
def convergence(exitingConfirmed: Set[UniqueAddress]): Boolean = {
// If another member in the data center that is UP or LEAVING and has not seen this gossip or is exiting
// convergence cannot be reached
def memberHinderingConvergenceExists =
members.exists(member ⇒
member.dataCenter == selfDc &&
convergenceMemberStatus(member.status) &&
!(latestGossip.seenByNode(member.uniqueAddress) || exitingConfirmed(member.uniqueAddress)))
// Find cluster members in the data center that are unreachable from other members of the data center
// excluding observations from members outside of the data center, that have status DOWN or is passed in as confirmed exiting.
val unreachableInDc = dcReachabilityExcludingDownedObservers.allUnreachableOrTerminated.collect {
case node if node != selfUniqueAddress && !exitingConfirmed(node) ⇒ latestGossip.member(node)
}
// unreachables outside of the data center or with status DOWN or EXITING does not affect convergence
val allUnreachablesCanBeIgnored =
unreachableInDc.forall(unreachable ⇒ convergenceSkipUnreachableWithMemberStatus(unreachable.status))
allUnreachablesCanBeIgnored && !memberHinderingConvergenceExists
}
/**
* @return Reachability excluding observations from nodes outside of the data center, but including observed unreachable
* nodes outside of the data center
*/
lazy val dcReachability: Reachability =
overview.reachability.removeObservers(
members.collect { case m if m.dataCenter != selfDc ⇒ m.uniqueAddress })
/**
* @return reachability for data center nodes, with observations from outside the data center or from downed nodes filtered out
*/
lazy val dcReachabilityExcludingDownedObservers: Reachability = {
val membersToExclude = members.collect { case m if m.status == Down || m.dataCenter != selfDc ⇒ m.uniqueAddress }
overview.reachability.removeObservers(membersToExclude).remove(members.collect { case m if m.dataCenter != selfDc ⇒ m.uniqueAddress })
}
/**
* @return true if toAddress should be reachable from the fromDc in general, within a data center
* this means only caring about data center local observations, across data centers it
* means caring about all observations for the toAddress.
*/
def isReachableExcludingDownedObservers(toAddress: UniqueAddress): Boolean =
if (!latestGossip.hasMember(toAddress)) false
else {
val to = latestGossip.member(toAddress)
// if member is in the same data center, we ignore cross data center unreachability
if (selfDc == to.dataCenter) dcReachabilityExcludingDownedObservers.isReachable(toAddress)
// if not it is enough that any non-downed node observed it as unreachable
else latestGossip.reachabilityExcludingDownedObservers.isReachable(toAddress)
}
def dcMembers: SortedSet[Member] =
members.filter(_.dataCenter == selfDc)
def isLeader(node: UniqueAddress): Boolean =
leader.contains(node)
def leader: Option[UniqueAddress] =
leaderOf(members)
def roleLeader(role: String): Option[UniqueAddress] =
leaderOf(members.filter(_.hasRole(role)))
def leaderOf(mbrs: immutable.SortedSet[Member]): Option[UniqueAddress] = {
val reachability = dcReachability
val reachableMembersInDc =
if (reachability.isAllReachable) mbrs.filter(m ⇒ m.dataCenter == selfDc && m.status != Down)
else mbrs.filter(m ⇒
m.dataCenter == selfDc &&
m.status != Down &&
(reachability.isReachable(m.uniqueAddress) || m.uniqueAddress == selfUniqueAddress))
if (reachableMembersInDc.isEmpty) None
else reachableMembersInDc.find(m ⇒ leaderMemberStatus(m.status))
.orElse(Some(reachableMembersInDc.min(Member.leaderStatusOrdering)))
.map(_.uniqueAddress)
}
}

50
akka-cluster/src/test/scala/akka/cluster/ClusterDomainEventPublisherSpec.scala
View file

@ -19,6 +19,7 @@ import akka.testkit.ImplicitSender
import akka.actor.ActorRef import akka.actor.ActorRef
import akka.remote.RARP import akka.remote.RARP
import akka.testkit.TestProbe import akka.testkit.TestProbe
import akka.cluster.ClusterSettings.DefaultDataCenter
object ClusterDomainEventPublisherSpec { object ClusterDomainEventPublisherSpec {
val config = """ val config = """
@ -36,6 +37,7 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
else "akka.tcp" else "akka.tcp"
var publisher: ActorRef = _ var publisher: ActorRef = _
val aUp = TestMember(Address(protocol, "sys", "a", 2552), Up) val aUp = TestMember(Address(protocol, "sys", "a", 2552), Up)
val aLeaving = aUp.copy(status = Leaving) val aLeaving = aUp.copy(status = Leaving)
val aExiting = aLeaving.copy(status = Exiting) val aExiting = aLeaving.copy(status = Exiting)
@ -48,16 +50,27 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
val a51Up = TestMember(Address(protocol, "sys", "a", 2551), Up) val a51Up = TestMember(Address(protocol, "sys", "a", 2551), Up)
val dUp = TestMember(Address(protocol, "sys", "d", 2552), Up, Set("GRP")) val dUp = TestMember(Address(protocol, "sys", "d", 2552), Up, Set("GRP"))
val emptyMembershipState = MembershipState(Gossip.empty, aUp.uniqueAddress, DefaultDataCenter)
val g0 = Gossip(members = SortedSet(aUp)).seen(aUp.uniqueAddress) val g0 = Gossip(members = SortedSet(aUp)).seen(aUp.uniqueAddress)
val state0 = MembershipState(g0, aUp.uniqueAddress, DefaultDataCenter)
val g1 = Gossip(members = SortedSet(aUp, cJoining)).seen(aUp.uniqueAddress).seen(cJoining.uniqueAddress) val g1 = Gossip(members = SortedSet(aUp, cJoining)).seen(aUp.uniqueAddress).seen(cJoining.uniqueAddress)
val state1 = MembershipState(g1, aUp.uniqueAddress, DefaultDataCenter)
val g2 = Gossip(members = SortedSet(aUp, bExiting, cUp)).seen(aUp.uniqueAddress) val g2 = Gossip(members = SortedSet(aUp, bExiting, cUp)).seen(aUp.uniqueAddress)
val state2 = MembershipState(g2, aUp.uniqueAddress, DefaultDataCenter)
val g3 = g2.seen(bExiting.uniqueAddress).seen(cUp.uniqueAddress) val g3 = g2.seen(bExiting.uniqueAddress).seen(cUp.uniqueAddress)
val state3 = MembershipState(g3, aUp.uniqueAddress, DefaultDataCenter)
val g4 = Gossip(members = SortedSet(a51Up, aUp, bExiting, cUp)).seen(aUp.uniqueAddress) val g4 = Gossip(members = SortedSet(a51Up, aUp, bExiting, cUp)).seen(aUp.uniqueAddress)
val state4 = MembershipState(g4, aUp.uniqueAddress, DefaultDataCenter)
val g5 = Gossip(members = SortedSet(a51Up, aUp, bExiting, cUp)).seen(aUp.uniqueAddress).seen(bExiting.uniqueAddress).seen(cUp.uniqueAddress).seen(a51Up.uniqueAddress) val g5 = Gossip(members = SortedSet(a51Up, aUp, bExiting, cUp)).seen(aUp.uniqueAddress).seen(bExiting.uniqueAddress).seen(cUp.uniqueAddress).seen(a51Up.uniqueAddress)
val state5 = MembershipState(g5, aUp.uniqueAddress, DefaultDataCenter)
val g6 = Gossip(members = SortedSet(aLeaving, bExiting, cUp)).seen(aUp.uniqueAddress) val g6 = Gossip(members = SortedSet(aLeaving, bExiting, cUp)).seen(aUp.uniqueAddress)
val state6 = MembershipState(g6, aUp.uniqueAddress, DefaultDataCenter)
val g7 = Gossip(members = SortedSet(aExiting, bExiting, cUp)).seen(aUp.uniqueAddress) val g7 = Gossip(members = SortedSet(aExiting, bExiting, cUp)).seen(aUp.uniqueAddress)
val state7 = MembershipState(g7, aUp.uniqueAddress, DefaultDataCenter)
val g8 = Gossip(members = SortedSet(aUp, bExiting, cUp, dUp), overview = GossipOverview(reachability = val g8 = Gossip(members = SortedSet(aUp, bExiting, cUp, dUp), overview = GossipOverview(reachability =
Reachability.empty.unreachable(aUp.uniqueAddress, dUp.uniqueAddress))).seen(aUp.uniqueAddress) Reachability.empty.unreachable(aUp.uniqueAddress, dUp.uniqueAddress))).seen(aUp.uniqueAddress)
val state8 = MembershipState(g8, aUp.uniqueAddress, DefaultDataCenter)
// created in beforeEach // created in beforeEach
var memberSubscriber: TestProbe = _ var memberSubscriber: TestProbe = _
@ -69,7 +82,7 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
system.eventStream.subscribe(memberSubscriber.ref, ClusterShuttingDown.getClass) system.eventStream.subscribe(memberSubscriber.ref, ClusterShuttingDown.getClass)
publisher = system.actorOf(Props[ClusterDomainEventPublisher]) publisher = system.actorOf(Props[ClusterDomainEventPublisher])
publisher ! PublishChanges(g0) publisher ! PublishChanges(state0)
memberSubscriber.expectMsg(MemberUp(aUp)) memberSubscriber.expectMsg(MemberUp(aUp))
memberSubscriber.expectMsg(LeaderChanged(Some(aUp.address))) memberSubscriber.expectMsg(LeaderChanged(Some(aUp.address)))
} }
@ -77,19 +90,19 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
"ClusterDomainEventPublisher" must { "ClusterDomainEventPublisher" must {
"publish MemberJoined" in { "publish MemberJoined" in {
publisher ! PublishChanges(g1) publisher ! PublishChanges(state1)
memberSubscriber.expectMsg(MemberJoined(cJoining)) memberSubscriber.expectMsg(MemberJoined(cJoining))
} }
"publish MemberUp" in { "publish MemberUp" in {
publisher ! PublishChanges(g2) publisher ! PublishChanges(state2)
publisher ! PublishChanges(g3) publisher ! PublishChanges(state3)
memberSubscriber.expectMsg(MemberExited(bExiting)) memberSubscriber.expectMsg(MemberExited(bExiting))
memberSubscriber.expectMsg(MemberUp(cUp)) memberSubscriber.expectMsg(MemberUp(cUp))
} }
"publish leader changed" in { "publish leader changed" in {
publisher ! PublishChanges(g4) publisher ! PublishChanges(state4)
memberSubscriber.expectMsg(MemberUp(a51Up)) memberSubscriber.expectMsg(MemberUp(a51Up))
memberSubscriber.expectMsg(MemberExited(bExiting)) memberSubscriber.expectMsg(MemberExited(bExiting))
memberSubscriber.expectMsg(MemberUp(cUp)) memberSubscriber.expectMsg(MemberUp(cUp))
@ -98,17 +111,17 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
} }
"publish leader changed when old leader leaves and is removed" in { "publish leader changed when old leader leaves and is removed" in {
publisher ! PublishChanges(g3) publisher ! PublishChanges(state3)
memberSubscriber.expectMsg(MemberExited(bExiting)) memberSubscriber.expectMsg(MemberExited(bExiting))
memberSubscriber.expectMsg(MemberUp(cUp)) memberSubscriber.expectMsg(MemberUp(cUp))
publisher ! PublishChanges(g6) publisher ! PublishChanges(state6)
memberSubscriber.expectMsg(MemberLeft(aLeaving)) memberSubscriber.expectMsg(MemberLeft(aLeaving))
publisher ! PublishChanges(g7) publisher ! PublishChanges(state7)
memberSubscriber.expectMsg(MemberExited(aExiting)) memberSubscriber.expectMsg(MemberExited(aExiting))
memberSubscriber.expectMsg(LeaderChanged(Some(cUp.address))) memberSubscriber.expectMsg(LeaderChanged(Some(cUp.address)))
memberSubscriber.expectNoMsg(500 millis) memberSubscriber.expectNoMsg(500 millis)
// at the removed member a an empty gossip is the last thing // at the removed member a an empty gossip is the last thing
publisher ! PublishChanges(Gossip.empty) publisher ! PublishChanges(emptyMembershipState)
memberSubscriber.expectMsg(MemberRemoved(aRemoved, Exiting)) memberSubscriber.expectMsg(MemberRemoved(aRemoved, Exiting))
memberSubscriber.expectMsg(MemberRemoved(bRemoved, Exiting)) memberSubscriber.expectMsg(MemberRemoved(bRemoved, Exiting))
memberSubscriber.expectMsg(MemberRemoved(cRemoved, Up)) memberSubscriber.expectMsg(MemberRemoved(cRemoved, Up))
@ -116,13 +129,13 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
} }
"not publish leader changed when same leader" in { "not publish leader changed when same leader" in {
publisher ! PublishChanges(g4) publisher ! PublishChanges(state4)
memberSubscriber.expectMsg(MemberUp(a51Up)) memberSubscriber.expectMsg(MemberUp(a51Up))
memberSubscriber.expectMsg(MemberExited(bExiting)) memberSubscriber.expectMsg(MemberExited(bExiting))
memberSubscriber.expectMsg(MemberUp(cUp)) memberSubscriber.expectMsg(MemberUp(cUp))
memberSubscriber.expectMsg(LeaderChanged(Some(a51Up.address))) memberSubscriber.expectMsg(LeaderChanged(Some(a51Up.address)))
publisher ! PublishChanges(g5) publisher ! PublishChanges(state5)
memberSubscriber.expectNoMsg(500 millis) memberSubscriber.expectNoMsg(500 millis)
} }
@ -130,12 +143,11 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
val subscriber = TestProbe() val subscriber = TestProbe()
publisher ! Subscribe(subscriber.ref, InitialStateAsSnapshot, Set(classOf[RoleLeaderChanged])) publisher ! Subscribe(subscriber.ref, InitialStateAsSnapshot, Set(classOf[RoleLeaderChanged]))
subscriber.expectMsgType[CurrentClusterState] subscriber.expectMsgType[CurrentClusterState]
publisher ! PublishChanges(Gossip(members = SortedSet(cJoining, dUp))) publisher ! PublishChanges(MembershipState(Gossip(members = SortedSet(cJoining, dUp)), dUp.uniqueAddress, DefaultDataCenter))
subscriber.expectMsgAllOf( subscriber.expectMsgAllOf(
RoleLeaderChanged("GRP", Some(dUp.address)), RoleLeaderChanged("GRP", Some(dUp.address)),
RoleLeaderChanged(ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter, Some(dUp.address)) RoleLeaderChanged(ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter, Some(dUp.address)))
) publisher ! PublishChanges(MembershipState(Gossip(members = SortedSet(cUp, dUp)), dUp.uniqueAddress, DefaultDataCenter))
publisher ! PublishChanges(Gossip(members = SortedSet(cUp, dUp)))
subscriber.expectMsg(RoleLeaderChanged("GRP", Some(cUp.address))) subscriber.expectMsg(RoleLeaderChanged("GRP", Some(cUp.address)))
} }
@ -150,7 +162,7 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
"send events corresponding to current state when subscribe" in { "send events corresponding to current state when subscribe" in {
val subscriber = TestProbe() val subscriber = TestProbe()
publisher ! PublishChanges(g8) publisher ! PublishChanges(state8)
publisher ! Subscribe(subscriber.ref, InitialStateAsEvents, Set(classOf[MemberEvent], classOf[ReachabilityEvent])) publisher ! Subscribe(subscriber.ref, InitialStateAsEvents, Set(classOf[MemberEvent], classOf[ReachabilityEvent]))
subscriber.receiveN(4).toSet should be(Set(MemberUp(aUp), MemberUp(cUp), MemberUp(dUp), MemberExited(bExiting))) subscriber.receiveN(4).toSet should be(Set(MemberUp(aUp), MemberUp(cUp), MemberUp(dUp), MemberExited(bExiting)))
subscriber.expectMsg(UnreachableMember(dUp)) subscriber.expectMsg(UnreachableMember(dUp))
@ -162,7 +174,7 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
publisher ! Subscribe(subscriber.ref, InitialStateAsSnapshot, Set(classOf[MemberEvent])) publisher ! Subscribe(subscriber.ref, InitialStateAsSnapshot, Set(classOf[MemberEvent]))
subscriber.expectMsgType[CurrentClusterState] subscriber.expectMsgType[CurrentClusterState]
publisher ! Unsubscribe(subscriber.ref, Some(classOf[MemberEvent])) publisher ! Unsubscribe(subscriber.ref, Some(classOf[MemberEvent]))
publisher ! PublishChanges(g3) publisher ! PublishChanges(state3)
subscriber.expectNoMsg(500 millis) subscriber.expectNoMsg(500 millis)
// but memberSubscriber is still subscriber // but memberSubscriber is still subscriber
memberSubscriber.expectMsg(MemberExited(bExiting)) memberSubscriber.expectMsg(MemberExited(bExiting))
@ -173,10 +185,10 @@ class ClusterDomainEventPublisherSpec extends AkkaSpec(ClusterDomainEventPublish
val subscriber = TestProbe() val subscriber = TestProbe()
publisher ! Subscribe(subscriber.ref, InitialStateAsSnapshot, Set(classOf[SeenChanged])) publisher ! Subscribe(subscriber.ref, InitialStateAsSnapshot, Set(classOf[SeenChanged]))
subscriber.expectMsgType[CurrentClusterState] subscriber.expectMsgType[CurrentClusterState]
publisher ! PublishChanges(g2) publisher ! PublishChanges(state2)
subscriber.expectMsgType[SeenChanged] subscriber.expectMsgType[SeenChanged]
subscriber.expectNoMsg(500 millis) subscriber.expectNoMsg(500 millis)
publisher ! PublishChanges(g3) publisher ! PublishChanges(state3)
subscriber.expectMsgType[SeenChanged] subscriber.expectMsgType[SeenChanged]
subscriber.expectNoMsg(500 millis) subscriber.expectNoMsg(500 millis)
} }

76
akka-cluster/src/test/scala/akka/cluster/ClusterDomainEventSpec.scala
View file

@ -38,30 +38,33 @@ class ClusterDomainEventSpec extends WordSpec with Matchers {
private[cluster] def converge(gossip: Gossip): (Gossip, Set[UniqueAddress]) = private[cluster] def converge(gossip: Gossip): (Gossip, Set[UniqueAddress]) =
((gossip, Set.empty[UniqueAddress]) /: gossip.members) { case ((gs, as), m) ⇒ (gs.seen(m.uniqueAddress), as + m.uniqueAddress) } ((gossip, Set.empty[UniqueAddress]) /: gossip.members) { case ((gs, as), m) ⇒ (gs.seen(m.uniqueAddress), as + m.uniqueAddress) }
private def state(g: Gossip): MembershipState =
MembershipState(g, selfDummyAddress, ClusterSettings.DefaultDataCenter)
"Domain events" must { "Domain events" must {
"be empty for the same gossip" in { "be empty for the same gossip" in {
val g1 = Gossip(members = SortedSet(aUp)) val g1 = Gossip(members = SortedSet(aUp))
diffUnreachable(g1, g1, selfDummyAddress) should ===(Seq.empty) diffUnreachable(state(g1), state(g1)) should ===(Seq.empty)
} }
"be produced for new members" in { "be produced for new members" in {
val (g1, _) = converge(Gossip(members = SortedSet(aUp))) val (g1, _) = converge(Gossip(members = SortedSet(aUp)))
val (g2, s2) = converge(Gossip(members = SortedSet(aUp, bUp, eJoining))) val (g2, s2) = converge(Gossip(members = SortedSet(aUp, bUp, eJoining)))
diffMemberEvents(g1, g2) should ===(Seq(MemberUp(bUp), MemberJoined(eJoining))) diffMemberEvents(state(g1), state(g2)) should ===(Seq(MemberUp(bUp), MemberJoined(eJoining)))
diffUnreachable(g1, g2, selfDummyAddress) should ===(Seq.empty) diffUnreachable(state(g1), state(g2)) should ===(Seq.empty)
diffSeen(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address)))) diffSeen(state(g1), state(g2)) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address))))
} }
"be produced for changed status of members" in { "be produced for changed status of members" in {
val (g1, _) = converge(Gossip(members = SortedSet(aJoining, bUp, cUp))) val (g1, _) = converge(Gossip(members = SortedSet(aJoining, bUp, cUp)))
val (g2, s2) = converge(Gossip(members = SortedSet(aUp, bUp, cLeaving, eJoining))) val (g2, s2) = converge(Gossip(members = SortedSet(aUp, bUp, cLeaving, eJoining)))
diffMemberEvents(g1, g2) should ===(Seq(MemberUp(aUp), MemberLeft(cLeaving), MemberJoined(eJoining))) diffMemberEvents(state(g1), state(g2)) should ===(Seq(MemberUp(aUp), MemberLeft(cLeaving), MemberJoined(eJoining)))
diffUnreachable(g1, g2, selfDummyAddress) should ===(Seq.empty) diffUnreachable(state(g1), state(g2)) should ===(Seq.empty)
diffSeen(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address)))) diffSeen(state(g1), state(g2)) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address))))
} }
"be produced for members in unreachable" in { "be produced for members in unreachable" in {
@ -73,10 +76,13 @@ class ClusterDomainEventSpec extends WordSpec with Matchers {
unreachable(aUp.uniqueAddress, bDown.uniqueAddress) unreachable(aUp.uniqueAddress, bDown.uniqueAddress)
val g2 = Gossip(members = SortedSet(aUp, cUp, bDown, eDown), overview = GossipOverview(reachability = reachability2)) val g2 = Gossip(members = SortedSet(aUp, cUp, bDown, eDown), overview = GossipOverview(reachability = reachability2))
diffUnreachable(g1, g2, selfDummyAddress) should ===(Seq(UnreachableMember(bDown))) diffUnreachable(state(g1), state(g2)) should ===(Seq(UnreachableMember(bDown)))
// never include self member in unreachable // never include self member in unreachable
diffUnreachable(g1, g2, bDown.uniqueAddress) should ===(Seq())
diffSeen(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===(Seq.empty) diffUnreachable(
MembershipState(g1, bDown.uniqueAddress, ClusterSettings.DefaultDataCenter),
MembershipState(g2, bDown.uniqueAddress, ClusterSettings.DefaultDataCenter)) should ===(Seq())
diffSeen(state(g1), state(g2)) should ===(Seq.empty)
} }
"be produced for members becoming reachable after unreachable" in { "be produced for members becoming reachable after unreachable" in {
@ -90,50 +96,54 @@ class ClusterDomainEventSpec extends WordSpec with Matchers {
reachable(aUp.uniqueAddress, bUp.uniqueAddress) reachable(aUp.uniqueAddress, bUp.uniqueAddress)
val g2 = Gossip(members = SortedSet(aUp, cUp, bUp, eUp), overview = GossipOverview(reachability = reachability2)) val g2 = Gossip(members = SortedSet(aUp, cUp, bUp, eUp), overview = GossipOverview(reachability = reachability2))
diffUnreachable(g1, g2, selfDummyAddress) should ===(Seq(UnreachableMember(cUp))) diffUnreachable(state(g1), state(g2)) should ===(Seq(UnreachableMember(cUp)))
// never include self member in unreachable // never include self member in unreachable
diffUnreachable(g1, g2, cUp.uniqueAddress) should ===(Seq()) diffUnreachable(
diffReachable(g1, g2, selfDummyAddress) should ===(Seq(ReachableMember(bUp))) MembershipState(g1, cUp.uniqueAddress, ClusterSettings.DefaultDataCenter),
MembershipState(g2, cUp.uniqueAddress, ClusterSettings.DefaultDataCenter)) should ===(Seq())
diffReachable(state(g1), state(g2)) should ===(Seq(ReachableMember(bUp)))
// never include self member in reachable // never include self member in reachable
diffReachable(g1, g2, bUp.uniqueAddress) should ===(Seq()) diffReachable(
MembershipState(g1, bUp.uniqueAddress, ClusterSettings.DefaultDataCenter),
MembershipState(g2, bUp.uniqueAddress, ClusterSettings.DefaultDataCenter)) should ===(Seq())
} }
"be produced for removed members" in { "be produced for removed members" in {
val (g1, _) = converge(Gossip(members = SortedSet(aUp, dExiting))) val (g1, _) = converge(Gossip(members = SortedSet(aUp, dExiting)))
val (g2, s2) = converge(Gossip(members = SortedSet(aUp))) val (g2, s2) = converge(Gossip(members = SortedSet(aUp)))
diffMemberEvents(g1, g2) should ===(Seq(MemberRemoved(dRemoved, Exiting))) diffMemberEvents(state(g1), state(g2)) should ===(Seq(MemberRemoved(dRemoved, Exiting)))
diffUnreachable(g1, g2, selfDummyAddress) should ===(Seq.empty) diffUnreachable(state(g1), state(g2)) should ===(Seq.empty)
diffSeen(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address)))) diffSeen(state(g1), state(g2)) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address))))
} }
"be produced for convergence changes" in { "be produced for convergence changes" in {
val g1 = Gossip(members = SortedSet(aUp, bUp, eJoining)).seen(aUp.uniqueAddress).seen(bUp.uniqueAddress).seen(eJoining.uniqueAddress) 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) val g2 = Gossip(members = SortedSet(aUp, bUp, eJoining)).seen(aUp.uniqueAddress).seen(bUp.uniqueAddress)
diffMemberEvents(g1, g2) should ===(Seq.empty) diffMemberEvents(state(g1), state(g2)) should ===(Seq.empty)
diffUnreachable(g1, g2, selfDummyAddress) should ===(Seq.empty) diffUnreachable(state(g1), state(g2)) should ===(Seq.empty)
diffSeen(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===(Seq(SeenChanged(convergence = true, seenBy = Set(aUp.address, bUp.address)))) diffSeen(state(g1), state(g2)) should ===(Seq(SeenChanged(convergence = true, seenBy = Set(aUp.address, bUp.address))))
diffMemberEvents(g2, g1) should ===(Seq.empty) diffMemberEvents(state(g2), state(g1)) should ===(Seq.empty)
diffUnreachable(g2, g1, selfDummyAddress) should ===(Seq.empty) diffUnreachable(state(g2), state(g1)) should ===(Seq.empty)
diffSeen(ClusterSettings.DefaultDataCenter, g2, g1, selfDummyAddress) should ===(Seq(SeenChanged(convergence = true, seenBy = Set(aUp.address, bUp.address, eJoining.address)))) diffSeen(state(g2), state(g1)) should ===(Seq(SeenChanged(convergence = true, seenBy = Set(aUp.address, bUp.address, eJoining.address))))
} }
"be produced for leader changes" in { "be produced for leader changes" in {
val (g1, _) = converge(Gossip(members = SortedSet(aUp, bUp, eJoining))) val (g1, _) = converge(Gossip(members = SortedSet(aUp, bUp, eJoining)))
val (g2, s2) = converge(Gossip(members = SortedSet(bUp, eJoining))) val (g2, s2) = converge(Gossip(members = SortedSet(bUp, eJoining)))
diffMemberEvents(g1, g2) should ===(Seq(MemberRemoved(aRemoved, Up))) diffMemberEvents(state(g1), state(g2)) should ===(Seq(MemberRemoved(aRemoved, Up)))
diffUnreachable(g1, g2, selfDummyAddress) should ===(Seq.empty) diffUnreachable(state(g1), state(g2)) should ===(Seq.empty)
diffSeen(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address)))) diffSeen(state(g1), state(g2)) should ===(Seq(SeenChanged(convergence = true, seenBy = s2.map(_.address))))
diffLeader(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===(Seq(LeaderChanged(Some(bUp.address)))) diffLeader(state(g1), state(g2)) should ===(Seq(LeaderChanged(Some(bUp.address))))
} }
"be produced for role leader changes in the same data center" in { "be produced for role leader changes in the same data center" in {
val g0 = Gossip.empty val g0 = Gossip.empty
val g1 = Gossip(members = SortedSet(aUp, bUp, cUp, dLeaving, eJoining)) val g1 = Gossip(members = SortedSet(aUp, bUp, cUp, dLeaving, eJoining))
val g2 = Gossip(members = SortedSet(bUp, cUp, dExiting, eJoining)) val g2 = Gossip(members = SortedSet(bUp, cUp, dExiting, eJoining))
diffRolesLeader(ClusterSettings.DefaultDataCenter, g0, g1, selfDummyAddress) should ===( diffRolesLeader(state(g0), state(g1)) should ===(
Set( Set(
// since this role is implicitly added // since this role is implicitly added
RoleLeaderChanged(ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter, Some(aUp.address)), RoleLeaderChanged(ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter, Some(aUp.address)),
@ -143,7 +153,7 @@ class ClusterDomainEventSpec extends WordSpec with Matchers {
RoleLeaderChanged("DD", Some(dLeaving.address)), RoleLeaderChanged("DD", Some(dLeaving.address)),
RoleLeaderChanged("DE", Some(dLeaving.address)), RoleLeaderChanged("DE", Some(dLeaving.address)),
RoleLeaderChanged("EE", Some(eUp.address)))) RoleLeaderChanged("EE", Some(eUp.address))))
diffRolesLeader(ClusterSettings.DefaultDataCenter, g1, g2, selfDummyAddress) should ===( diffRolesLeader(state(g1), state(g2)) should ===(
Set( Set(
RoleLeaderChanged(ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter, Some(bUp.address)), RoleLeaderChanged(ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter, Some(bUp.address)),
RoleLeaderChanged("AA", None), RoleLeaderChanged("AA", None),
@ -153,10 +163,14 @@ class ClusterDomainEventSpec extends WordSpec with Matchers {
"not be produced for role leader changes in other data centers" in { "not be produced for role leader changes in other data centers" in {
val g0 = Gossip.empty val g0 = Gossip.empty
val s0 = state(g0).copy(selfDc = "dc2")
val g1 = Gossip(members = SortedSet(aUp, bUp, cUp, dLeaving, eJoining)) val g1 = Gossip(members = SortedSet(aUp, bUp, cUp, dLeaving, eJoining))
val s1 = state(g1).copy(selfDc = "dc2")
val g2 = Gossip(members = SortedSet(bUp, cUp, dExiting, eJoining)) val g2 = Gossip(members = SortedSet(bUp, cUp, dExiting, eJoining))
diffRolesLeader("dc2", g0, g1, selfDummyAddress) should ===(Set.empty) val s2 = state(g2).copy(selfDc = "dc2")
diffRolesLeader("dc2", g1, g2, selfDummyAddress) should ===(Set.empty)
diffRolesLeader(s0, s1) should ===(Set.empty)
diffRolesLeader(s1, s2) should ===(Set.empty)
} }
} }
} }

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

@ -7,6 +7,7 @@ package akka.cluster
import org.scalatest.WordSpec import org.scalatest.WordSpec
import org.scalatest.Matchers import org.scalatest.Matchers
import akka.actor.Address import akka.actor.Address
import akka.cluster.ClusterSettings.DataCenter
import akka.cluster.ClusterSettings.DefaultDataCenter import akka.cluster.ClusterSettings.DefaultDataCenter
import scala.collection.immutable.SortedSet import scala.collection.immutable.SortedSet
@ -33,6 +34,9 @@ class GossipSpec extends WordSpec with Matchers {
val dc2d1 = TestMember(Address("akka.tcp", "sys", "d", 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) val dc2d2 = TestMember(dc2d1.address, status = Down, roles = Set.empty, dataCenter = dc2d1.dataCenter)
private def state(g: Gossip, selfMember: Member = a1): MembershipState =
MembershipState(g, selfMember.uniqueAddress, selfMember.dataCenter)
"A Gossip" must { "A Gossip" must {
"have correct test setup" in { "have correct test setup" in {
@ -41,40 +45,40 @@ class GossipSpec extends WordSpec with Matchers {
} }
"reach convergence when it's empty" in { "reach convergence when it's empty" in {
Gossip.empty.convergence(DefaultDataCenter, a1.uniqueAddress, Set.empty) should ===(true) state(Gossip.empty).convergence(Set.empty) should ===(true)
} }
"reach convergence for one node" in { "reach convergence for one node" in {
val g1 = Gossip(members = SortedSet(a1)).seen(a1.uniqueAddress) val g1 = Gossip(members = SortedSet(a1)).seen(a1.uniqueAddress)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set.empty) should ===(true) state(g1).convergence(Set.empty) should ===(true)
} }
"not reach convergence until all have seen version" in { "not reach convergence until all have seen version" in {
val g1 = Gossip(members = SortedSet(a1, b1)).seen(a1.uniqueAddress) val g1 = Gossip(members = SortedSet(a1, b1)).seen(a1.uniqueAddress)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set.empty) should ===(false) state(g1).convergence(Set.empty) should ===(false)
} }
"reach convergence for two nodes" in { "reach convergence for two nodes" in {
val g1 = Gossip(members = SortedSet(a1, b1)).seen(a1.uniqueAddress).seen(b1.uniqueAddress) val g1 = Gossip(members = SortedSet(a1, b1)).seen(a1.uniqueAddress).seen(b1.uniqueAddress)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set.empty) should ===(true) state(g1).convergence(Set.empty) should ===(true)
} }
"reach convergence, skipping joining" in { "reach convergence, skipping joining" in {
// e1 is joining // e1 is joining
val g1 = Gossip(members = SortedSet(a1, b1, e1)).seen(a1.uniqueAddress).seen(b1.uniqueAddress) val g1 = Gossip(members = SortedSet(a1, b1, e1)).seen(a1.uniqueAddress).seen(b1.uniqueAddress)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set.empty) should ===(true) state(g1).convergence(Set.empty) should ===(true)
} }
"reach convergence, skipping down" in { "reach convergence, skipping down" in {
// e3 is down // e3 is down
val g1 = Gossip(members = SortedSet(a1, b1, e3)).seen(a1.uniqueAddress).seen(b1.uniqueAddress) val g1 = Gossip(members = SortedSet(a1, b1, e3)).seen(a1.uniqueAddress).seen(b1.uniqueAddress)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set.empty) should ===(true) state(g1).convergence(Set.empty) should ===(true)
} }
"reach convergence, skipping Leaving with exitingConfirmed" in { "reach convergence, skipping Leaving with exitingConfirmed" in {
// c1 is Leaving // c1 is Leaving
val g1 = Gossip(members = SortedSet(a1, b1, c1)).seen(a1.uniqueAddress).seen(b1.uniqueAddress) val g1 = Gossip(members = SortedSet(a1, b1, c1)).seen(a1.uniqueAddress).seen(b1.uniqueAddress)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set(c1.uniqueAddress)) should ===(true) state(g1).convergence(Set(c1.uniqueAddress)) should ===(true)
} }
"reach convergence, skipping unreachable Leaving with exitingConfirmed" in { "reach convergence, skipping unreachable Leaving with exitingConfirmed" in {
@ -82,16 +86,16 @@ class GossipSpec extends WordSpec with Matchers {
val r1 = Reachability.empty.unreachable(b1.uniqueAddress, c1.uniqueAddress) val r1 = Reachability.empty.unreachable(b1.uniqueAddress, c1.uniqueAddress)
val g1 = Gossip(members = SortedSet(a1, b1, c1), overview = GossipOverview(reachability = r1)) val g1 = Gossip(members = SortedSet(a1, b1, c1), overview = GossipOverview(reachability = r1))
.seen(a1.uniqueAddress).seen(b1.uniqueAddress) .seen(a1.uniqueAddress).seen(b1.uniqueAddress)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set(c1.uniqueAddress)) should ===(true) state(g1).convergence(Set(c1.uniqueAddress)) should ===(true)
} }
"not reach convergence when unreachable" in { "not reach convergence when unreachable" in {
val r1 = Reachability.empty.unreachable(b1.uniqueAddress, a1.uniqueAddress) val r1 = Reachability.empty.unreachable(b1.uniqueAddress, a1.uniqueAddress)
val g1 = (Gossip(members = SortedSet(a1, b1), overview = GossipOverview(reachability = r1))) val g1 = (Gossip(members = SortedSet(a1, b1), overview = GossipOverview(reachability = r1)))
.seen(a1.uniqueAddress).seen(b1.uniqueAddress) .seen(a1.uniqueAddress).seen(b1.uniqueAddress)
g1.convergence(DefaultDataCenter, b1.uniqueAddress, Set.empty) should ===(false) state(g1, b1).convergence(Set.empty) should ===(false)
// but from a1's point of view (it knows that itself is not unreachable) // but from a1's point of view (it knows that itself is not unreachable)
g1.convergence(DefaultDataCenter, a1.uniqueAddress, Set.empty) should ===(true) state(g1).convergence(Set.empty) should ===(true)
} }
"reach convergence when downed node has observed unreachable" in { "reach convergence when downed node has observed unreachable" in {
@ -99,7 +103,7 @@ class GossipSpec extends WordSpec with Matchers {
val r1 = Reachability.empty.unreachable(e3.uniqueAddress, a1.uniqueAddress) val r1 = Reachability.empty.unreachable(e3.uniqueAddress, a1.uniqueAddress)
val g1 = (Gossip(members = SortedSet(a1, b1, e3), overview = GossipOverview(reachability = r1))) val g1 = (Gossip(members = SortedSet(a1, b1, e3), overview = GossipOverview(reachability = r1)))
.seen(a1.uniqueAddress).seen(b1.uniqueAddress).seen(e3.uniqueAddress) .seen(a1.uniqueAddress).seen(b1.uniqueAddress).seen(e3.uniqueAddress)
g1.convergence(DefaultDataCenter, b1.uniqueAddress, Set.empty) should ===(true) state(g1, b1).convergence(Set.empty) should ===(true)
} }
"merge members by status priority" in { "merge members by status priority" in {
@ -146,37 +150,33 @@ class GossipSpec extends WordSpec with Matchers {
} }
"have leader as first member based on ordering, except Exiting status" in { "have leader as first member based on ordering, except Exiting status" in {
Gossip(members = SortedSet(c2, e2)).dcLeader(DefaultDataCenter, c2.uniqueAddress) should ===(Some(c2.uniqueAddress)) state(Gossip(members = SortedSet(c2, e2)), c2).leader should ===(Some(c2.uniqueAddress))
Gossip(members = SortedSet(c3, e2)).dcLeader(DefaultDataCenter, c3.uniqueAddress) should ===(Some(e2.uniqueAddress)) state(Gossip(members = SortedSet(c3, e2)), c3).leader should ===(Some(e2.uniqueAddress))
Gossip(members = SortedSet(c3)).dcLeader(DefaultDataCenter, c3.uniqueAddress) should ===(Some(c3.uniqueAddress)) state(Gossip(members = SortedSet(c3)), c3).leader should ===(Some(c3.uniqueAddress))
} }
"have leader as first reachable member based on ordering" in { "have leader as first reachable member based on ordering" in {
val r1 = Reachability.empty.unreachable(e2.uniqueAddress, c2.uniqueAddress) val r1 = Reachability.empty.unreachable(e2.uniqueAddress, c2.uniqueAddress)
val g1 = Gossip(members = SortedSet(c2, e2), overview = GossipOverview(reachability = r1)) val g1 = Gossip(members = SortedSet(c2, e2), overview = GossipOverview(reachability = r1))
g1.dcLeader(DefaultDataCenter, e2.uniqueAddress) should ===(Some(e2.uniqueAddress)) state(g1, e2).leader should ===(Some(e2.uniqueAddress))
// but when c2 is selfUniqueAddress // but when c2 is selfUniqueAddress
g1.dcLeader(DefaultDataCenter, c2.uniqueAddress) should ===(Some(c2.uniqueAddress)) state(g1, c2).leader should ===(Some(c2.uniqueAddress))
} }
"not have Down member as leader" in { "not have Down member as leader" in {
Gossip(members = SortedSet(e3)).dcLeader(DefaultDataCenter, e3.uniqueAddress) should ===(None) state(Gossip(members = SortedSet(e3)), e3).leader should ===(None)
} }
"have a leader per data center" in { "have a leader per data center" in {
val g1 = Gossip(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d1)) val g1 = Gossip(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d1))
// everybodys point of view is dc1a1 being leader of dc1 // dc1a1 being leader of dc1
g1.dcLeader("dc1", dc1a1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g1, dc1a1).leader should ===(Some(dc1a1.uniqueAddress))
g1.dcLeader("dc1", dc1b1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g1, dc1b1).leader should ===(Some(dc1a1.uniqueAddress))
g1.dcLeader("dc1", dc2c1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress))
g1.dcLeader("dc1", dc2d1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress))
// and dc2c1 being leader of dc2 // and dc2c1 being leader of dc2
g1.dcLeader("dc2", dc1a1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g1, dc2c1).leader should ===(Some(dc2c1.uniqueAddress))
g1.dcLeader("dc2", dc1b1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g1, dc2d1).leader should ===(Some(dc2c1.uniqueAddress))
g1.dcLeader("dc2", dc2c1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress))
g1.dcLeader("dc2", dc2d1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress))
} }
"merge seen table correctly" in { "merge seen table correctly" in {
@ -218,11 +218,11 @@ class GossipSpec extends WordSpec with Matchers {
.seen(dc1b1.uniqueAddress) .seen(dc1b1.uniqueAddress)
.seen(dc2c1.uniqueAddress) .seen(dc2c1.uniqueAddress)
.seen(dc2d1.uniqueAddress) .seen(dc2d1.uniqueAddress)
g.dcLeader("dc1", dc1a1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g, dc1a1).leader should ===(Some(dc1a1.uniqueAddress))
g.convergence("dc1", dc1a1.uniqueAddress, Set.empty) should ===(true) state(g, dc1a1).convergence(Set.empty) should ===(true)
g.dcLeader("dc2", dc2c1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g, dc2c1).leader should ===(Some(dc2c1.uniqueAddress))
g.convergence("dc2", dc2c1.uniqueAddress, Set.empty) should ===(true) state(g, dc2c1).convergence(Set.empty) should ===(true)
} }
"reach convergence per data center even if members of another data center has not seen the gossip" in { "reach convergence per data center even if members of another data center has not seen the gossip" in {
@ -233,12 +233,12 @@ class GossipSpec extends WordSpec with Matchers {
// dc2d1 has not seen the gossip // dc2d1 has not seen the gossip
// so dc1 can reach convergence // so dc1 can reach convergence
g.dcLeader("dc1", dc1a1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g, dc1a1).leader should ===(Some(dc1a1.uniqueAddress))
g.convergence("dc1", dc1a1.uniqueAddress, Set.empty) should ===(true) state(g, dc1a1).convergence(Set.empty) should ===(true)
// but dc2 cannot // but dc2 cannot
g.dcLeader("dc2", dc2c1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g, dc2c1).leader should ===(Some(dc2c1.uniqueAddress))
g.convergence("dc2", dc2c1.uniqueAddress, Set.empty) should ===(false) state(g, dc2c1).convergence(Set.empty) should ===(false)
} }
"reach convergence per data center even if another data center contains unreachable" in { "reach convergence per data center even if another data center contains unreachable" in {
@ -251,12 +251,12 @@ class GossipSpec extends WordSpec with Matchers {
.seen(dc2d1.uniqueAddress) .seen(dc2d1.uniqueAddress)
// this data center doesn't care about dc2 having reachability problems and can reach convergence // this data center doesn't care about dc2 having reachability problems and can reach convergence
g.dcLeader("dc1", dc1a1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g, dc1a1).leader should ===(Some(dc1a1.uniqueAddress))
g.convergence("dc1", dc1a1.uniqueAddress, Set.empty) should ===(true) state(g, dc1a1).convergence(Set.empty) should ===(true)
// this data center is cannot reach convergence because of unreachability within the data center // this data center is cannot reach convergence because of unreachability within the data center
g.dcLeader("dc2", dc2c1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g, dc2c1).leader should ===(Some(dc2c1.uniqueAddress))
g.convergence("dc2", dc2c1.uniqueAddress, Set.empty) should ===(false) state(g, dc2c1).convergence(Set.empty) should ===(false)
} }
"reach convergence per data center even if there is unreachable nodes in another data center" in { "reach convergence per data center even if there is unreachable nodes in another data center" in {
@ -271,11 +271,11 @@ class GossipSpec extends WordSpec with Matchers {
.seen(dc2d1.uniqueAddress) .seen(dc2d1.uniqueAddress)
// neither data center is affected by the inter data center unreachability as far as convergence goes // neither data center is affected by the inter data center unreachability as far as convergence goes
g.dcLeader("dc1", dc1a1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g, dc1a1).leader should ===(Some(dc1a1.uniqueAddress))
g.convergence("dc1", dc1a1.uniqueAddress, Set.empty) should ===(true) state(g, dc1a1).convergence(Set.empty) should ===(true)
g.dcLeader("dc2", dc2c1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g, dc2c1).leader should ===(Some(dc2c1.uniqueAddress))
g.convergence("dc2", dc2c1.uniqueAddress, Set.empty) should ===(true) state(g, dc2c1).convergence(Set.empty) should ===(true)
} }
"ignore cross data center unreachability when determining inside of data center reachability" in { "ignore cross data center unreachability when determining inside of data center reachability" in {
@ -291,10 +291,10 @@ class GossipSpec extends WordSpec with Matchers {
g.isReachable(dc2c1.uniqueAddress, dc2d1.uniqueAddress) should ===(true) g.isReachable(dc2c1.uniqueAddress, dc2d1.uniqueAddress) should ===(true)
g.isReachable(dc2d1.uniqueAddress, dc2c1.uniqueAddress) should ===(true) g.isReachable(dc2d1.uniqueAddress, dc2c1.uniqueAddress) should ===(true)
g.isReachableExcludingDownedObservers(dc1a1.dataCenter, dc1b1.uniqueAddress) should ===(true) state(g, dc1a1).isReachableExcludingDownedObservers(dc1b1.uniqueAddress) should ===(true)
g.isReachableExcludingDownedObservers(dc1b1.dataCenter, dc1a1.uniqueAddress) should ===(true) state(g, dc1b1).isReachableExcludingDownedObservers(dc1a1.uniqueAddress) should ===(true)
g.isReachableExcludingDownedObservers(dc2c1.dataCenter, dc2d1.uniqueAddress) should ===(true) state(g, dc2c1).isReachableExcludingDownedObservers(dc2d1.uniqueAddress) should ===(true)
g.isReachableExcludingDownedObservers(dc2d1.dataCenter, dc2c1.uniqueAddress) should ===(true) state(g, dc2d1).isReachableExcludingDownedObservers(dc2c1.uniqueAddress) should ===(true)
// between data centers it matters though // between data centers it matters though
g.isReachable(dc1a1.uniqueAddress, dc2c1.uniqueAddress) should ===(false) g.isReachable(dc1a1.uniqueAddress, dc2c1.uniqueAddress) should ===(false)
@ -304,22 +304,22 @@ class GossipSpec extends WordSpec with Matchers {
g.isReachable(dc2d1.uniqueAddress, dc1a1.uniqueAddress) should ===(true) g.isReachable(dc2d1.uniqueAddress, dc1a1.uniqueAddress) should ===(true)
// this one looks at all unreachable-entries for the to-address // this one looks at all unreachable-entries for the to-address
g.isReachableExcludingDownedObservers(dc1a1.dataCenter, dc2c1.uniqueAddress) should ===(false) state(g, dc1a1).isReachableExcludingDownedObservers(dc2c1.uniqueAddress) should ===(false)
g.isReachableExcludingDownedObservers(dc1b1.dataCenter, dc2c1.uniqueAddress) should ===(false) state(g, dc1b1).isReachableExcludingDownedObservers(dc2c1.uniqueAddress) should ===(false)
g.isReachableExcludingDownedObservers(dc2c1.dataCenter, dc1a1.uniqueAddress) should ===(false) state(g, dc2c1).isReachableExcludingDownedObservers(dc1a1.uniqueAddress) should ===(false)
g.isReachableExcludingDownedObservers(dc2d1.dataCenter, dc1a1.uniqueAddress) should ===(false) state(g, dc2d1).isReachableExcludingDownedObservers(dc1a1.uniqueAddress) should ===(false)
// between the two other nodes there is no unreachability // between the two other nodes there is no unreachability
g.isReachable(dc1b1.uniqueAddress, dc2d1.uniqueAddress) should ===(true) g.isReachable(dc1b1.uniqueAddress, dc2d1.uniqueAddress) should ===(true)
g.isReachable(dc2d1.uniqueAddress, dc1b1.uniqueAddress) should ===(true) g.isReachable(dc2d1.uniqueAddress, dc1b1.uniqueAddress) should ===(true)
g.isReachableExcludingDownedObservers(dc1b1.dataCenter, dc2d1.uniqueAddress) should ===(true) state(g, dc1b1).isReachableExcludingDownedObservers(dc2d1.uniqueAddress) should ===(true)
g.isReachableExcludingDownedObservers(dc2d1.dataCenter, dc1b1.uniqueAddress) should ===(true) state(g, dc2d1).isReachableExcludingDownedObservers(dc1b1.uniqueAddress) should ===(true)
} }
"not returning a downed data center leader" in { "not returning a downed data center leader" in {
val g = Gossip(members = SortedSet(dc1a1.copy(Down), dc1b1)) val g = Gossip(members = SortedSet(dc1a1.copy(Down), dc1b1))
g.leaderOf("dc1", g.members, dc1b1.uniqueAddress) should ===(Some(dc1b1.uniqueAddress)) state(g, dc1b1).leaderOf(g.members) should ===(Some(dc1b1.uniqueAddress))
} }
"ignore cross data center unreachability when determining data center leader" in { "ignore cross data center unreachability when determining data center leader" in {
@ -329,15 +329,11 @@ class GossipSpec extends WordSpec with Matchers {
val g = Gossip(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d1), overview = GossipOverview(reachability = r1)) val g = Gossip(members = SortedSet(dc1a1, dc1b1, dc2c1, dc2d1), overview = GossipOverview(reachability = r1))
g.leaderOf("dc1", g.members, dc1a1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g, dc1a1).leaderOf(g.members) should ===(Some(dc1a1.uniqueAddress))
g.leaderOf("dc1", g.members, dc1b1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress)) state(g, dc1b1).leaderOf(g.members) should ===(Some(dc1a1.uniqueAddress))
g.leaderOf("dc1", g.members, dc2c1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress))
g.leaderOf("dc1", g.members, dc2d1.uniqueAddress) should ===(Some(dc1a1.uniqueAddress))
g.leaderOf("dc2", g.members, dc1a1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g, dc2c1).leaderOf(g.members) should ===(Some(dc2c1.uniqueAddress))
g.leaderOf("dc2", g.members, dc1b1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress)) state(g, dc2d1).leaderOf(g.members) should ===(Some(dc2c1.uniqueAddress))
g.leaderOf("dc2", g.members, dc2c1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress))
g.leaderOf("dc2", g.members, dc2d1.uniqueAddress) should ===(Some(dc2c1.uniqueAddress))
} }
// TODO test coverage for when leaderOf returns None - I have not been able to figure it out // TODO test coverage for when leaderOf returns None - I have not been able to figure it out

40
project/MiMa.scala
View file

@ -1221,14 +1221,6 @@ object MiMa extends AutoPlugin {
// #22881 Make sure connections are aborted correctly on Windows // #22881 Make sure connections are aborted correctly on Windows
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.io.ChannelRegistration.cancel"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.io.ChannelRegistration.cancel"),
// #23231 multi-DC Sharding
ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.cluster.ddata.Replicator.leader"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.Replicator.receiveLeaderChanged"),
ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.cluster.ddata.Replicator.leader_="),
FilterAnyProblemStartingWith("akka.cluster.sharding.ClusterShardingGuardian"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.sharding.ShardRegion.proxyProps"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.sharding.ShardRegion.this"),
// #23144 recoverWithRetries cleanup // #23144 recoverWithRetries cleanup
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.fusing.RecoverWith.InfiniteRetries"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.fusing.RecoverWith.InfiniteRetries"),
@ -1237,23 +1229,29 @@ object MiMa extends AutoPlugin {
// #23023 added a new overload with implementation to trait, so old transport implementations compiled against // #23023 added a new overload with implementation to trait, so old transport implementations compiled against
// older versions will be missing the method. We accept that incompatibility for now. // older versions will be missing the method. We accept that incompatibility for now.
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.remote.transport.AssociationHandle.disassociate"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.remote.transport.AssociationHandle.disassociate")
),
"2.5.3" -> Seq(
// #23231 multi-DC Sharding
ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.cluster.ddata.Replicator.leader"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.Replicator.receiveLeaderChanged"),
ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.cluster.ddata.Replicator.leader_="),
FilterAnyProblemStartingWith("akka.cluster.sharding.ClusterShardingGuardian"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.sharding.ShardRegion.proxyProps"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.sharding.ShardRegion.this"),
// #23228 single leader per cluster data center // #23228 single leader per cluster data center
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.apply"), FilterAnyProblemStartingWith("akka.cluster.Gossip"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.copy"), FilterAnyProblemStartingWith("akka.cluster.ClusterCoreDaemon"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.this"), FilterAnyProblemStartingWith("akka.cluster.ClusterDomainEventPublisher"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.convergence"), FilterAnyProblemStartingWith("akka.cluster.InternalClusterAction"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.isLeader"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffReachable"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.leader"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.leaderOf"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.Gossip.roleLeader"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterCoreDaemon.NumberOfGossipsBeforeShutdownWhenLeaderExits"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterCoreDaemon.vclockName"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterCoreDaemon.MaxGossipsBeforeShuttingDownMyself"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffLeader"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffLeader"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffRolesLeader"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffRolesLeader"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffSeen"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffSeen"),
ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.cluster.ClusterEvent.diffReachability"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterEvent.diffUnreachable"),
ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.cluster.ClusterEvent.diffMemberEvents"),
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.cluster.protobuf.msg.ClusterMessages#GossipOrBuilder.getTombstonesCount"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.cluster.protobuf.msg.ClusterMessages#GossipOrBuilder.getTombstonesCount"),
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.cluster.protobuf.msg.ClusterMessages#GossipOrBuilder.getTombstones"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.cluster.protobuf.msg.ClusterMessages#GossipOrBuilder.getTombstones"),
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.cluster.protobuf.msg.ClusterMessages#GossipOrBuilder.getTombstonesList"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.cluster.protobuf.msg.ClusterMessages#GossipOrBuilder.getTombstonesList"),