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

Fix regression IllegalStateException: Can't find this Member, see #2117

Browse source 
* Gossiping to myself caused the failure detector to mark myself as unreachable
* Don't heartbeat for myself failure detector
This commit is contained in:
Patrik Nordwall 2012-05-31 17:19:49 +02:00
parent 9d1cbdc601
commit 13a93dfb50
2 changed files with 53 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -203,7 +203,7 @@ case class Gossip(
} }
/** /**
* Marks the gossip as seen by this node (remoteAddress) by updating the address entry in the 'gossip.overview.seen' * Marks the gossip as seen by this node (selfAddress) by updating the address entry in the 'gossip.overview.seen'
* Map with the VectorClock for the new gossip. * Map with the VectorClock for the new gossip.
*/ */
def seen(address: Address): Gossip = { def seen(address: Address): Gossip = {
@ -380,11 +380,11 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val remoteSettings = new RemoteSettings(system.settings.config, system.name) val remoteSettings = new RemoteSettings(system.settings.config, system.name)
val clusterSettings = new ClusterSettings(system.settings.config, system.name) val clusterSettings = new ClusterSettings(system.settings.config, system.name)
val remoteAddress = remote.transport.address val selfAddress = remote.transport.address
val failureDetector = new AccrualFailureDetector( val failureDetector = new AccrualFailureDetector(
system, remoteAddress, clusterSettings.FailureDetectorThreshold, clusterSettings.FailureDetectorMaxSampleSize) system, selfAddress, clusterSettings.FailureDetectorThreshold, clusterSettings.FailureDetectorMaxSampleSize)
private val vclockNode = VectorClock.Node(remoteAddress.toString) private val vclockNode = VectorClock.Node(selfAddress.toString)
private val periodicTasksInitialDelay = clusterSettings.PeriodicTasksInitialDelay private val periodicTasksInitialDelay = clusterSettings.PeriodicTasksInitialDelay
private val gossipFrequency = clusterSettings.GossipFrequency private val gossipFrequency = clusterSettings.GossipFrequency
@ -396,7 +396,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
private val autoDown = clusterSettings.AutoDown private val autoDown = clusterSettings.AutoDown
private val nrOfDeputyNodes = clusterSettings.NrOfDeputyNodes private val nrOfDeputyNodes = clusterSettings.NrOfDeputyNodes
private val nrOfGossipDaemons = clusterSettings.NrOfGossipDaemons private val nrOfGossipDaemons = clusterSettings.NrOfGossipDaemons
private val nodeToJoin: Option[Address] = clusterSettings.NodeToJoin filter (_ != remoteAddress) private val nodeToJoin: Option[Address] = clusterSettings.NodeToJoin filter (_ != selfAddress)
private val serialization = remote.serialization private val serialization = remote.serialization
@ -406,7 +406,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
private val mBeanServer = ManagementFactory.getPlatformMBeanServer private val mBeanServer = ManagementFactory.getPlatformMBeanServer
private val clusterMBeanName = new ObjectName("akka:type=Cluster") private val clusterMBeanName = new ObjectName("akka:type=Cluster")
log.info("Cluster Node [{}] - is starting up...", remoteAddress) log.info("Cluster Node [{}] - is starting up...", selfAddress)
// create superisor for daemons under path "/system/cluster" // create superisor for daemons under path "/system/cluster"
private val clusterDaemons = { private val clusterDaemons = {
@ -418,7 +418,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
} }
private val state = { private val state = {
val member = Member(remoteAddress, MemberStatus.Joining) val member = Member(selfAddress, MemberStatus.Joining)
val gossip = Gossip(members = SortedSet.empty[Member] + member) + vclockNode // add me as member and update my vector clock val gossip = Gossip(members = SortedSet.empty[Member] + member) + vclockNode // add me as member and update my vector clock
new AtomicReference[State](State(gossip)) new AtomicReference[State](State(gossip))
} }
@ -447,15 +447,15 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
createMBean() createMBean()
log.info("Cluster Node [{}] - has started up successfully", remoteAddress) log.info("Cluster Node [{}] - has started up successfully", selfAddress)
// ====================================================== // ======================================================
// ===================== PUBLIC API ===================== // ===================== PUBLIC API =====================
// ====================================================== // ======================================================
def self: Member = latestGossip.members def self: Member = latestGossip.members
.find(_.address == remoteAddress) .find(_.address == selfAddress)
.getOrElse(throw new IllegalStateException("Can't find 'this' Member (" + remoteAddress + ") in the cluster membership ring")) .getOrElse(throw new IllegalStateException("Can't find 'this' Member (" + selfAddress + ") in the cluster membership ring"))
/** /**
* Latest gossip. * Latest gossip.
@ -472,7 +472,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
*/ */
def isLeader: Boolean = { def isLeader: Boolean = {
val members = latestGossip.members val members = latestGossip.members
!members.isEmpty && (remoteAddress == members.head.address) members.nonEmpty && (selfAddress == members.head.address)
} }
/** /**
@ -502,7 +502,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
*/ */
def shutdown(): Unit = { def shutdown(): Unit = {
if (isRunning.compareAndSet(true, false)) { if (isRunning.compareAndSet(true, false)) {
log.info("Cluster Node [{}] - Shutting down cluster Node and cluster daemons...", remoteAddress) log.info("Cluster Node [{}] - Shutting down cluster Node and cluster daemons...", selfAddress)
gossipCanceller.cancel() gossipCanceller.cancel()
failureDetectorReaperCanceller.cancel() failureDetectorReaperCanceller.cancel()
leaderActionsCanceller.cancel() leaderActionsCanceller.cancel()
@ -543,8 +543,8 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
*/ */
def join(address: Address): Unit = { def join(address: Address): Unit = {
val connection = clusterCommandConnectionFor(address) val connection = clusterCommandConnectionFor(address)
val command = ClusterAction.Join(remoteAddress) val command = ClusterAction.Join(selfAddress)
log.info("Cluster Node [{}] - Trying to send JOIN to [{}] through connection [{}]", remoteAddress, address, connection) log.info("Cluster Node [{}] - Trying to send JOIN to [{}] through connection [{}]", selfAddress, address, connection)
connection ! command connection ! command
} }
@ -579,7 +579,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
*/ */
@tailrec @tailrec
private[cluster] final def joining(node: Address): Unit = { private[cluster] final def joining(node: Address): Unit = {
log.info("Cluster Node [{}] - Node [{}] is JOINING", remoteAddress, node) log.info("Cluster Node [{}] - Node [{}] is JOINING", selfAddress, node)
val localState = state.get val localState = state.get
val localGossip = localState.latestGossip val localGossip = localState.latestGossip
@ -595,13 +595,14 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val newGossip = localGossip copy (overview = newOverview, members = newMembers) val newGossip = localGossip copy (overview = newOverview, members = newMembers)
val versionedGossip = newGossip + vclockNode val versionedGossip = newGossip + vclockNode
val seenVersionedGossip = versionedGossip seen remoteAddress val seenVersionedGossip = versionedGossip seen selfAddress
val newState = localState copy (latestGossip = seenVersionedGossip) val newState = localState copy (latestGossip = seenVersionedGossip)
if (!state.compareAndSet(localState, newState)) joining(node) // recur if we failed update if (!state.compareAndSet(localState, newState)) joining(node) // recur if we failed update
else { else {
failureDetector heartbeat node // update heartbeat in failure detector if (node != selfAddress) failureDetector heartbeat node
if (convergence(newState.latestGossip).isDefined) { if (convergence(newState.latestGossip).isDefined) {
newState.memberMembershipChangeListeners foreach { _ notify newMembers } newState.memberMembershipChangeListeners foreach { _ notify newMembers }
} }
@ -612,28 +613,28 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
* State transition to UP. * State transition to UP.
*/ */
private[cluster] final def up(address: Address): Unit = { private[cluster] final def up(address: Address): Unit = {
log.info("Cluster Node [{}] - Marking node [{}] as UP", remoteAddress, address) log.info("Cluster Node [{}] - Marking node [{}] as UP", selfAddress, address)
} }
/** /**
* State transition to LEAVING. * State transition to LEAVING.
*/ */
private[cluster] final def leaving(address: Address): Unit = { private[cluster] final def leaving(address: Address): Unit = {
log.info("Cluster Node [{}] - Marking node [{}] as LEAVING", remoteAddress, address) log.info("Cluster Node [{}] - Marking node [{}] as LEAVING", selfAddress, address)
} }
/** /**
* State transition to EXITING. * State transition to EXITING.
*/ */
private[cluster] final def exiting(address: Address): Unit = { private[cluster] final def exiting(address: Address): Unit = {
log.info("Cluster Node [{}] - Marking node [{}] as EXITING", remoteAddress, address) log.info("Cluster Node [{}] - Marking node [{}] as EXITING", selfAddress, address)
} }
/** /**
* State transition to REMOVED. * State transition to REMOVED.
*/ */
private[cluster] final def removing(address: Address): Unit = { private[cluster] final def removing(address: Address): Unit = {
log.info("Cluster Node [{}] - Marking node [{}] as REMOVED", remoteAddress, address) log.info("Cluster Node [{}] - Marking node [{}] as REMOVED", selfAddress, address)
} }
/** /**
@ -658,7 +659,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
localMembers localMembers
.map { member ⇒ .map { member ⇒
if (member.address == address) { if (member.address == address) {
log.info("Cluster Node [{}] - Marking node [{}] as DOWN", remoteAddress, member.address) log.info("Cluster Node [{}] - Marking node [{}] as DOWN", selfAddress, member.address)
val newMember = member copy (status = MemberStatus.Down) val newMember = member copy (status = MemberStatus.Down)
downedMember = Some(newMember) downedMember = Some(newMember)
newMember newMember
@ -672,7 +673,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
.filter(_.status != MemberStatus.Down) // no need to DOWN members already DOWN .filter(_.status != MemberStatus.Down) // no need to DOWN members already DOWN
.map { member ⇒ .map { member ⇒
if (member.address == address) { if (member.address == address) {
log.info("Cluster Node [{}] - Marking unreachable node [{}] as DOWN", remoteAddress, member.address) log.info("Cluster Node [{}] - Marking unreachable node [{}] as DOWN", selfAddress, member.address)
member copy (status = MemberStatus.Down) member copy (status = MemberStatus.Down)
} else member } else member
} }
@ -691,7 +692,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val newOverview = localOverview copy (seen = newSeen, unreachable = newUnreachablePlusNewlyDownedMembers) // update gossip overview val newOverview = localOverview copy (seen = newSeen, unreachable = newUnreachablePlusNewlyDownedMembers) // update gossip overview
val newGossip = localGossip copy (overview = newOverview, members = newMembers) // update gossip val newGossip = localGossip copy (overview = newOverview, members = newMembers) // update gossip
val versionedGossip = newGossip + vclockNode val versionedGossip = newGossip + vclockNode
val newState = localState copy (latestGossip = versionedGossip seen remoteAddress) val newState = localState copy (latestGossip = versionedGossip seen selfAddress)
if (!state.compareAndSet(localState, newState)) downing(address) // recur if we fail the update if (!state.compareAndSet(localState, newState)) downing(address) // recur if we fail the update
else { else {
@ -730,14 +731,14 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
remoteGossip remoteGossip
} }
val newState = localState copy (latestGossip = winningGossip seen remoteAddress) val newState = localState copy (latestGossip = winningGossip seen selfAddress)
// if we won the race then update else try again // if we won the race then update else try again
if (!state.compareAndSet(localState, newState)) receive(sender, remoteGossip) // recur if we fail the update if (!state.compareAndSet(localState, newState)) receive(sender, remoteGossip) // recur if we fail the update
else { else {
log.debug("Cluster Node [{}] - Receiving gossip from [{}]", remoteAddress, sender.address) log.debug("Cluster Node [{}] - Receiving gossip from [{}]", selfAddress, sender.address)
failureDetector heartbeat sender.address // update heartbeat in failure detector if (sender.address != selfAddress) failureDetector heartbeat sender.address
if (convergence(newState.latestGossip).isDefined) { if (convergence(newState.latestGossip).isDefined) {
newState.memberMembershipChangeListeners foreach { _ notify newState.latestGossip.members } newState.memberMembershipChangeListeners foreach { _ notify newState.latestGossip.members }
@ -758,7 +759,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
* @return the updated new state with the new member status * @return the updated new state with the new member status
*/ */
private def switchMemberStatusTo(newStatus: MemberStatus, state: State): State = { private def switchMemberStatusTo(newStatus: MemberStatus, state: State): State = {
log.info("Cluster Node [{}] - Switching membership status to [{}]", remoteAddress, newStatus) log.info("Cluster Node [{}] - Switching membership status to [{}]", selfAddress, newStatus)
val localSelf = self val localSelf = self
@ -770,7 +771,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
// change my state in 'gossip.members' // change my state in 'gossip.members'
val newMembersSet = localMembers map { member ⇒ val newMembersSet = localMembers map { member ⇒
if (member.address == remoteAddress) newSelf if (member.address == selfAddress) newSelf
else member else member
} }
@ -780,7 +781,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
// version my changes // version my changes
val versionedGossip = newGossip + vclockNode val versionedGossip = newGossip + vclockNode
val seenVersionedGossip = versionedGossip seen remoteAddress val seenVersionedGossip = versionedGossip seen selfAddress
state copy (latestGossip = seenVersionedGossip) state copy (latestGossip = seenVersionedGossip)
} }
@ -790,7 +791,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
*/ */
private def gossipTo(address: Address): Unit = { private def gossipTo(address: Address): Unit = {
val connection = clusterGossipConnectionFor(address) val connection = clusterGossipConnectionFor(address)
log.debug("Cluster Node [{}] - Gossiping to [{}]", remoteAddress, connection) log.debug("Cluster Node [{}] - Gossiping to [{}]", selfAddress, connection)
connection ! GossipEnvelope(self, latestGossip) connection ! GossipEnvelope(self, latestGossip)
} }
@ -800,10 +801,10 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
* @return 'true' if it gossiped to a "deputy" member. * @return 'true' if it gossiped to a "deputy" member.
*/ */
private def gossipToRandomNodeOf(addresses: Iterable[Address]): Boolean = { private def gossipToRandomNodeOf(addresses: Iterable[Address]): Boolean = {
log.debug("Cluster Node [{}] - Selecting random node to gossip to [{}]", remoteAddress, addresses.mkString(", ")) log.debug("Cluster Node [{}] - Selecting random node to gossip to [{}]", selfAddress, addresses.mkString(", "))
if (addresses.isEmpty) false if (addresses.isEmpty) false
else { else {
val peers = addresses filter (_ != remoteAddress) // filter out myself val peers = addresses filter (_ != selfAddress) // filter out myself
val peer = selectRandomNode(peers) val peer = selectRandomNode(peers)
gossipTo(peer) gossipTo(peer)
deputyNodes exists (peer == _) deputyNodes exists (peer == _)
@ -819,10 +820,10 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
if (isSingletonCluster(localState)) { if (isSingletonCluster(localState)) {
// gossip to myself // gossip to myself
// TODO could perhaps be optimized, no need to gossip to myself when Up? // TODO could perhaps be optimized, no need to gossip to myself when Up?
gossipTo(remoteAddress) gossipTo(selfAddress)
} else if (isAvailable(localState)) { } else if (isAvailable(localState)) {
log.debug("Cluster Node [{}] - Initiating new round of gossip", remoteAddress) log.debug("Cluster Node [{}] - Initiating new round of gossip", selfAddress)
val localGossip = localState.latestGossip val localGossip = localState.latestGossip
val localMembers = localGossip.members val localMembers = localGossip.members
@ -842,7 +843,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
// 3. gossip to a deputy nodes for facilitating partition healing // 3. gossip to a deputy nodes for facilitating partition healing
val deputies = deputyNodes val deputies = deputyNodes
if ((!gossipedToDeputy || localMembersSize < 1) && !deputies.isEmpty) { if ((!gossipedToDeputy || localMembersSize < 1) && deputies.nonEmpty) {
if (localMembersSize == 0) gossipToRandomNodeOf(deputies) if (localMembersSize == 0) gossipToRandomNodeOf(deputies)
else { else {
val probability = 1.0 / localMembersSize + localUnreachableSize val probability = 1.0 / localMembersSize + localUnreachableSize
@ -870,7 +871,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val newlyDetectedUnreachableMembers = localMembers filterNot { member ⇒ failureDetector.isAvailable(member.address) } val newlyDetectedUnreachableMembers = localMembers filterNot { member ⇒ failureDetector.isAvailable(member.address) }
if (!newlyDetectedUnreachableMembers.isEmpty) { // we have newly detected members marked as unavailable if (newlyDetectedUnreachableMembers.nonEmpty) { // we have newly detected members marked as unavailable
val newMembers = localMembers diff newlyDetectedUnreachableMembers val newMembers = localMembers diff newlyDetectedUnreachableMembers
val newUnreachableMembers: Set[Member] = localUnreachableMembers ++ newlyDetectedUnreachableMembers val newUnreachableMembers: Set[Member] = localUnreachableMembers ++ newlyDetectedUnreachableMembers
@ -880,14 +881,14 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
// updating vclock and 'seen' table // updating vclock and 'seen' table
val versionedGossip = newGossip + vclockNode val versionedGossip = newGossip + vclockNode
val seenVersionedGossip = versionedGossip seen remoteAddress val seenVersionedGossip = versionedGossip seen selfAddress
val newState = localState copy (latestGossip = seenVersionedGossip) val newState = localState copy (latestGossip = seenVersionedGossip)
// if we won the race then update else try again // if we won the race then update else try again
if (!state.compareAndSet(localState, newState)) reapUnreachableMembers() // recur if (!state.compareAndSet(localState, newState)) reapUnreachableMembers() // recur
else { else {
log.info("Cluster Node [{}] - Marking node(s) as UNREACHABLE [{}]", remoteAddress, newlyDetectedUnreachableMembers.mkString(", ")) log.info("Cluster Node [{}] - Marking node(s) as UNREACHABLE [{}]", selfAddress, newlyDetectedUnreachableMembers.mkString(", "))
if (convergence(newState.latestGossip).isDefined) { if (convergence(newState.latestGossip).isDefined) {
newState.memberMembershipChangeListeners foreach { _ notify newMembers } newState.memberMembershipChangeListeners foreach { _ notify newMembers }
@ -906,7 +907,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val localGossip = localState.latestGossip val localGossip = localState.latestGossip
val localMembers = localGossip.members val localMembers = localGossip.members
val isLeader = !localMembers.isEmpty && (remoteAddress == localMembers.head.address) val isLeader = localMembers.nonEmpty && (selfAddress == localMembers.head.address)
if (isLeader && isAvailable(localState)) { if (isLeader && isAvailable(localState)) {
// only run the leader actions if we are the LEADER and available // only run the leader actions if we are the LEADER and available
@ -932,14 +933,14 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
localMembers map { member ⇒ localMembers map { member ⇒
// 1. Move JOINING => UP // 1. Move JOINING => UP
if (member.status == MemberStatus.Joining) { if (member.status == MemberStatus.Joining) {
log.info("Cluster Node [{}] - Leader is moving node [{}] from JOINING to UP", remoteAddress, member.address) log.info("Cluster Node [{}] - Leader is moving node [{}] from JOINING to UP", selfAddress, member.address)
hasChangedState = true hasChangedState = true
member copy (status = MemberStatus.Up) member copy (status = MemberStatus.Up)
} else member } else member
} map { member ⇒ } map { member ⇒
// 2. Move EXITING => REMOVED // 2. Move EXITING => REMOVED
if (member.status == MemberStatus.Exiting) { if (member.status == MemberStatus.Exiting) {
log.info("Cluster Node [{}] - Leader is moving node [{}] from EXITING to REMOVED", remoteAddress, member.address) log.info("Cluster Node [{}] - Leader is moving node [{}] from EXITING to REMOVED", selfAddress, member.address)
hasChangedState = true hasChangedState = true
member copy (status = MemberStatus.Removed) member copy (status = MemberStatus.Removed)
} else member } else member
@ -955,7 +956,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
localUnreachableMembers localUnreachableMembers
.filter(_.status != MemberStatus.Down) // no need to DOWN members already DOWN .filter(_.status != MemberStatus.Down) // no need to DOWN members already DOWN
.map { member ⇒ .map { member ⇒
log.info("Cluster Node [{}] - Leader is marking unreachable node [{}] as DOWN", remoteAddress, member.address) log.info("Cluster Node [{}] - Leader is marking unreachable node [{}] as DOWN", selfAddress, member.address)
hasChangedState = true hasChangedState = true
member copy (status = MemberStatus.Down) member copy (status = MemberStatus.Down)
} }
@ -974,7 +975,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val versionedGossip = newGossip + vclockNode val versionedGossip = newGossip + vclockNode
// 5. Updating the 'seen' table // 5. Updating the 'seen' table
val seenVersionedGossip = versionedGossip seen remoteAddress val seenVersionedGossip = versionedGossip seen selfAddress
val newState = localState copy (latestGossip = seenVersionedGossip) val newState = localState copy (latestGossip = seenVersionedGossip)
@ -1009,7 +1010,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val views = Set.empty[VectorClock] ++ seen.values val views = Set.empty[VectorClock] ++ seen.values
if (views.size == 1) { if (views.size == 1) {
log.debug("Cluster Node [{}] - Cluster convergence reached", remoteAddress) log.debug("Cluster Node [{}] - Cluster convergence reached", selfAddress)
Some(gossip) Some(gossip)
} else None } else None
} else None } else None
@ -1022,7 +1023,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val localOverview = localGossip.overview val localOverview = localGossip.overview
val localMembers = localGossip.members val localMembers = localGossip.members
val localUnreachableMembers = localOverview.unreachable val localUnreachableMembers = localOverview.unreachable
val isUnreachable = localUnreachableMembers exists { _.address == remoteAddress } val isUnreachable = localUnreachableMembers exists { _.address == selfAddress }
val hasUnavailableMemberStatus = localMembers exists { m ⇒ (m == self) && MemberStatus.isUnavailable(m.status) } val hasUnavailableMemberStatus = localMembers exists { m ⇒ (m == self) && MemberStatus.isUnavailable(m.status) }
isUnreachable || hasUnavailableMemberStatus isUnreachable || hasUnavailableMemberStatus
} }
@ -1030,7 +1031,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
/** /**
* Looks up and returns the local cluster command connection. * Looks up and returns the local cluster command connection.
*/ */
private def clusterCommandDaemon = system.actorFor(RootActorPath(remoteAddress) / "system" / "cluster" / "commands") private def clusterCommandDaemon = system.actorFor(RootActorPath(selfAddress) / "system" / "cluster" / "commands")
/** /**
* Looks up and returns the remote cluster command connection for the specific address. * Looks up and returns the remote cluster command connection for the specific address.
@ -1045,7 +1046,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
/** /**
* Gets an Iterable with the addresses of a all the 'deputy' nodes - excluding this node if part of the group. * Gets an Iterable with the addresses of a all the 'deputy' nodes - excluding this node if part of the group.
*/ */
private def deputyNodes: Iterable[Address] = state.get.latestGossip.members.toIterable map (_.address) drop 1 take nrOfDeputyNodes filter (_ != remoteAddress) private def deputyNodes: Iterable[Address] = state.get.latestGossip.members.toIterable map (_.address) drop 1 take nrOfDeputyNodes filter (_ != selfAddress)
private def selectRandomNode(addresses: Iterable[Address]): Address = addresses.toSeq(ThreadLocalRandom.current nextInt addresses.size) private def selectRandomNode(addresses: Iterable[Address]): Address = addresses.toSeq(ThreadLocalRandom.current nextInt addresses.size)
@ -1074,8 +1075,8 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
val unreachable = gossip.overview.unreachable val unreachable = gossip.overview.unreachable
val metaData = gossip.meta val metaData = gossip.meta
"\nMembers:\n\t" + gossip.members.mkString("\n\t") + "\nMembers:\n\t" + gossip.members.mkString("\n\t") +
{ if (!unreachable.isEmpty) "\nUnreachable:\n\t" + unreachable.mkString("\n\t") else "" } + { if (unreachable.nonEmpty) "\nUnreachable:\n\t" + unreachable.mkString("\n\t") else "" } +
{ if (!metaData.isEmpty) "\nMeta Data:\t" + metaData.toString else "" } { if (metaData.nonEmpty) "\nMeta Data:\t" + metaData.toString else "" }
} }
def getMemberStatus: String = clusterNode.status.toString def getMemberStatus: String = clusterNode.status.toString
@ -1100,7 +1101,7 @@ class Cluster(system: ExtendedActorSystem) extends Extension { clusterNode ⇒
def shutdown() = clusterNode.shutdown() def shutdown() = clusterNode.shutdown()
} }
log.info("Cluster Node [{}] - registering cluster JMX MBean [{}]", remoteAddress, clusterMBeanName) log.info("Cluster Node [{}] - registering cluster JMX MBean [{}]", selfAddress, clusterMBeanName)
try { try {
mBeanServer.registerMBean(mbean, clusterMBeanName) mBeanServer.registerMBean(mbean, clusterMBeanName)
} catch { } catch {

1
akka-cluster/src/multi-jvm/scala/akka/cluster/NodeStartupSpec.scala
View file

@ -38,6 +38,7 @@ abstract class NodeStartupSpec extends MultiNodeSpec(NodeStartupMultiJvmSpec) wi
runOn(first) { runOn(first) {
awaitCond(cluster.isSingletonCluster) awaitCond(cluster.isSingletonCluster)
awaitUpConvergence(numberOfMembers = 1) awaitUpConvergence(numberOfMembers = 1)
cluster.isLeader must be(true)
} }
} }
} }