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

MultiDcSplitBain: only subscribe to unreachable after split

Browse source 
Test would fail picking up the reachable from the previous unsplit
as it is a new probe.

Also change barrierCounter to split/unsplit so easier to see
where the failure is on a barrier fail
This commit is contained in:
Christopher Batey 2018-01-30 08:57:37 +00:00
parent 78b5045165
commit 5658d6e77a
2 changed files with 13 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -596,9 +596,6 @@ private[cluster] final class ClusterDomainEventPublisher extends Actor with Acto
}
def publishDiff(oldState: MembershipState, newState: MembershipState, pub: AnyRef ⇒ Unit): Unit = {
def inSameDc(reachabilityEvent: ReachabilityEvent): Boolean =
reachabilityEvent.member.dataCenter == selfDc
diffMemberEvents(oldState, newState) foreach pub
diffUnreachable(oldState, newState) foreach pub
diffReachable(oldState, newState) foreach pub

28
akka-cluster/src/multi-jvm/scala/akka/cluster/MultiDcSplitBrainSpec.scala
View file

@ -68,17 +68,18 @@ abstract class MultiDcSplitBrainSpec
val dc1 = List(first, second)
val dc2 = List(third, fourth, fifth)
var barrierCounter = 0
var splits = 0
var unsplits = 0
def splitDataCenters(doNotVerify: Set[RoleName]): Unit = {
splits += 1
val memberNodes = (dc1 ++ dc2).filterNot(doNotVerify)
val probe = TestProbe()
runOn(memberNodes: _*) {
cluster.subscribe(probe.ref, classOf[DataCenterReachabilityEvent])
cluster.subscribe(probe.ref, classOf[UnreachableDataCenter])
probe.expectMsgType[CurrentClusterState]
}
enterBarrier(s"split-$barrierCounter")
barrierCounter += 1
enterBarrier(s"split-$splits")
runOn(first) {
for (dc1Node ← dc1; dc2Node ← dc2) {
@ -86,8 +87,7 @@ abstract class MultiDcSplitBrainSpec
}
}
enterBarrier(s"after-split-$barrierCounter")
barrierCounter += 1
enterBarrier(s"after-split-$splits")
runOn(memberNodes: _*) {
probe.expectMsgType[UnreachableDataCenter](15.seconds)
@ -104,19 +104,18 @@ abstract class MultiDcSplitBrainSpec
}
cluster.state.unreachable should ===(Set.empty)
}
enterBarrier(s"after-split-verified-$barrierCounter")
barrierCounter += 1
enterBarrier(s"after-split-verified-$splits")
}
def unsplitDataCenters(notMembers: Set[RoleName]): Unit = {
unsplits += 1
val memberNodes = (dc1 ++ dc2).filterNot(notMembers)
val probe = TestProbe()
runOn(memberNodes: _*) {
cluster.subscribe(probe.ref, classOf[ReachableDataCenter])
probe.expectMsgType[CurrentClusterState]
}
enterBarrier(s"unsplit-$barrierCounter")
barrierCounter += 1
enterBarrier(s"unsplit-$unsplits")
runOn(first) {
for (dc1Node ← dc1; dc2Node ← dc2) {
@ -124,19 +123,18 @@ abstract class MultiDcSplitBrainSpec
}
}
enterBarrier(s"after-unsplit-$barrierCounter")
barrierCounter += 1
enterBarrier(s"after-unsplit-$unsplits")
runOn(memberNodes: _*) {
probe.expectMsgType[ReachableDataCenter](25.seconds)
system.log.debug("Reachable data center received")
cluster.unsubscribe(probe.ref)
awaitAssert {
cluster.state.unreachableDataCenters should ===(Set.empty)
system.log.debug("Cluster state: {}", cluster.state)
}
}
enterBarrier(s"after-unsplit-verified-$barrierCounter")
barrierCounter += 1
enterBarrier(s"after-unsplit-verified-$unsplits")
}
"A cluster with multiple data centers" must {