You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2026/10/09 07:39:25 UTC
(pekko) branch main updated: test: cover that harmless quarantine does not trigger DownSelfQuarantinedByRemote (#3550)
This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new a5cb8e9653 test: cover that harmless quarantine does not trigger DownSelfQuarantinedByRemote (#3550)
a5cb8e9653 is described below
commit a5cb8e96538663637fdaa6d85e5116d5f5d7371d
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Fri Oct 9 08:39:18 2026 +0100
test: cover that harmless quarantine does not trigger DownSelfQuarantinedByRemote (#3550)
Motivation:
Port of akka/akka-core#32117 (backport of akka/akka#32050, issue akka/akka#31095). That fix stopped `InboundQuarantineCheck` replying with a `Quarantined` control message for later inbound messages from a harmlessly quarantined (gracefully shut down) node, which made the remote node down itself via `DownSelfQuarantinedByRemote` - notably during a network partition.
Pekko already stops propagating harmless quarantines by default: #1555 added `pekko.remote.artery.propagate-harmless-quarantine-events = off`, and `QuarantinedTimestamp.harmless` / `newQuarantined(harmless)` already exist, so no MiMa filters are needed. What was missing was the multi-jvm coverage of the end-to-end cluster scenario.
Modification:
- `DowningWhenOtherHasQuarantinedThisActorSystemSpec`: ported the Akka rewrite. Quarantines `second` directly via `RARP(system).provider.quarantine(...)` instead of relying on split brain resolver timing, asserts `second` receives `ThisActorSystemQuarantinedEvent` and terminates within 5 s (shorter than `stable-after`, so it is not normal downing), adds a `fourth` node and a new case "not be triggered by another node shutting down during network partition" - the actual #31095 scenario [...]
- `Association` / `InboundQuarantineCheck`: log at info when a `Quarantined` control message is sent, and mention "Harmless quarantine." in the harmless quarantine log message, matching Akka. Log-only.
- `TestContext`: pass `harmless = false` explicitly, as in Akka.
Result:
Multi-jvm coverage of the `DownSelfQuarantinedByRemote` scenarios, including harmless quarantine during a network partition. No behavioural change beyond log messages.
Tests:
- `sbt "project cluster" "MultiJvm/testOnly org.apache.pekko.cluster.DowningWhenOtherHasQuarantinedThisActorSystem"` - 4 nodes, 4 tests each, all passed
- `sbt "remote/testOnly org.apache.pekko.remote.artery.HarmlessQuarantineSpec org.apache.pekko.remote.artery.OutboundIdleShutdownSpec org.apache.pekko.remote.artery.RemoteMessageSerializationSpec"` - 14 passed
- native scalafmt run on the changed files
- MiMa not run: no signature changes
References:
Refs #1555 - port of akka/akka-core#32117 (the `InboundQuarantineCheck` fix was already present). Akka source is Apache-2.0 licensed as of 2023-09-20.
---
...henOtherHasQuarantinedThisActorSystemSpec.scala | 116 ++++++++++++++-------
.../apache/pekko/remote/artery/Association.scala | 5 +-
.../remote/artery/InboundQuarantineCheck.scala | 4 +-
.../apache/pekko/remote/artery/TestContext.scala | 2 +-
4 files changed, 85 insertions(+), 42 deletions(-)
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/DowningWhenOtherHasQuarantinedThisActorSystemSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/DowningWhenOtherHasQuarantinedThisActorSystemSpec.scala
index 0e882f2bd9..22eb1ac0f1 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/DowningWhenOtherHasQuarantinedThisActorSystemSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/DowningWhenOtherHasQuarantinedThisActorSystemSpec.scala
@@ -16,14 +16,18 @@ package org.apache.pekko.cluster
import scala.concurrent.duration._
import org.apache.pekko
+import pekko.actor.ActorIdentity
import pekko.actor.ActorRef
import pekko.actor.Identify
import pekko.actor.RootActorPath
+import pekko.remote.RARP
import pekko.remote.artery.ArterySettings
import pekko.remote.artery.ThisActorSystemQuarantinedEvent
import pekko.remote.testkit.MultiNodeConfig
import pekko.remote.transport.ThrottlerTransportAdapter
+import pekko.testkit.EventFilter
import pekko.testkit.LongRunningTest
+import pekko.testkit.TestEvent.Mute
import com.typesafe.config.ConfigFactory
@@ -31,6 +35,7 @@ object DowningWhenOtherHasQuarantinedThisActorSystemSpec extends MultiNodeConfig
val first = role("first")
val second = role("second")
val third = role("third")
+ val fourth = role("fourth")
commonConfig(
debugConfig(on = false)
@@ -39,15 +44,9 @@ object DowningWhenOtherHasQuarantinedThisActorSystemSpec extends MultiNodeConfig
ConfigFactory.parseString("""
pekko.remote.artery.enabled = on
pekko.cluster.downing-provider-class = "org.apache.pekko.cluster.sbr.SplitBrainResolverProvider"
- # speed up decision
- pekko.cluster.split-brain-resolver.stable-after = 5s
+ pekko.cluster.split-brain-resolver.stable-after = 10s
""")))
- // exaggerate the timing issue by ,making the second node decide slower
- // this is to more consistently repeat the scenario where the other side completes downing
- // while the isolated part still has not made a decision and then see quarantined connections from the other nodes
- nodeConfig(second)(ConfigFactory.parseString("pekko.cluster.split-brain-resolver.stable-after = 15s"))
-
testTransport(on = true)
}
@@ -57,11 +56,16 @@ class DowningWhenOtherHasQuarantinedThisActorSystemMultiJvmNode2
extends DowningWhenOtherHasQuarantinedThisActorSystemSpec
class DowningWhenOtherHasQuarantinedThisActorSystemMultiJvmNode3
extends DowningWhenOtherHasQuarantinedThisActorSystemSpec
+class DowningWhenOtherHasQuarantinedThisActorSystemMultiJvmNode4
+ extends DowningWhenOtherHasQuarantinedThisActorSystemSpec
abstract class DowningWhenOtherHasQuarantinedThisActorSystemSpec
extends MultiNodeClusterSpec(DowningWhenOtherHasQuarantinedThisActorSystemSpec) {
import DowningWhenOtherHasQuarantinedThisActorSystemSpec._
+ muteDeadLetters(classOf[ActorIdentity])()
+ system.eventStream.publish(Mute(EventFilter.info(pattern = ".*Ignoring received gossip from unknown.*")))
+
"Cluster node downed by other" must {
if (!ArterySettings(system.settings.config.getConfig("pekko.remote.artery")).Enabled) {
@@ -71,43 +75,23 @@ abstract class DowningWhenOtherHasQuarantinedThisActorSystemSpec
}
"join cluster" taggedAs LongRunningTest in {
- awaitClusterUp(first, second, third)
+ awaitClusterUp(first, second, third, fourth)
enterBarrier("after-1")
}
- "down itself" taggedAs LongRunningTest in {
- runOn(first) {
- testConductor.blackhole(first, second, ThrottlerTransportAdapter.Direction.Both).await
- testConductor.blackhole(third, second, ThrottlerTransportAdapter.Direction.Both).await
- }
- enterBarrier("blackhole")
-
- within(15.seconds) {
- runOn(first) {
- awaitAssert {
- cluster.state.unreachable.map(_.address) should ===(Set(address(second)))
- }
- awaitAssert {
- // second downed and removed
- cluster.state.members.map(_.address) should ===(Set(address(first), address(third)))
- }
- }
- runOn(second) {
- awaitAssert {
- cluster.state.unreachable.map(_.address) should ===(Set(address(first), address(third)))
- }
- }
+ "down itself with DownSelfQuarantinedByRemote when other has quarantined" taggedAs LongRunningTest in {
+ runOn(first, second) {
+ system.eventStream.subscribe(testActor, classOf[ThisActorSystemQuarantinedEvent])
}
- enterBarrier("down-second")
-
runOn(first) {
- testConductor.passThrough(first, second, ThrottlerTransportAdapter.Direction.Both).await
- testConductor.passThrough(third, second, ThrottlerTransportAdapter.Direction.Both).await
+ val secondUniqueAddress = cluster.state.members.find(_.address == address(second)).get.uniqueAddress
+ RARP(system).provider
+ .quarantine(secondUniqueAddress.address, Some(secondUniqueAddress.longUid), "Quarantine from test")
}
- enterBarrier("pass-through")
+ enterBarrier("quarantined")
runOn(second) {
- within(10.seconds) {
+ within(5.seconds) { // this is shorter than split-brain-resolver.stable-after, so it's not normal downing
awaitAssert {
// try to ping first (Cluster Heartbeat messages will not trigger the Quarantine message)
system.actorSelection(RootActorPath(first) / "user").tell(Identify(None), ActorRef.noSender)
@@ -115,6 +99,21 @@ abstract class DowningWhenOtherHasQuarantinedThisActorSystemSpec
cluster.isTerminated should ===(true)
}
}
+ expectMsgType[ThisActorSystemQuarantinedEvent]
+ }
+ enterBarrier("second-shutdown")
+
+ runOn(first) {
+ expectNoMessage(1.second) // no ThisActorSystemQuarantinedEvent
+ }
+ enterBarrier("wait")
+
+ runOn(first) {
+ val sel = system.actorSelection(RootActorPath(second) / "user")
+ (1 to 15).foreach { _ =>
+ sel.tell(Identify(None), ActorRef.noSender) // try to ping second
+ expectNoMessage(200.millis) // no ThisActorSystemQuarantinedEvent
+ }
}
enterBarrier("after-2")
@@ -130,15 +129,56 @@ abstract class DowningWhenOtherHasQuarantinedThisActorSystemSpec
cluster.shutdown()
}
+ runOn(first) {
+ expectNoMessage(1.second) // no ThisActorSystemQuarantinedEvent
+ }
+ enterBarrier("wait")
+
runOn(first) {
val sel = system.actorSelection(RootActorPath(third) / "user")
- (1 to 25).foreach { _ =>
+ (1 to 15).foreach { _ =>
sel.tell(Identify(None), ActorRef.noSender) // try to ping third
expectNoMessage(200.millis) // no ThisActorSystemQuarantinedEvent
}
}
- enterBarrier("after-2")
+ enterBarrier("after-3")
+ }
+
+ "not be triggered by another node shutting down during network partition" taggedAs LongRunningTest in {
+ runOn(first) {
+ system.eventStream.subscribe(testActor, classOf[ThisActorSystemQuarantinedEvent])
+ }
+ enterBarrier("subscribing")
+
+ runOn(first) {
+ testConductor.blackhole(first, fourth, ThrottlerTransportAdapter.Direction.Both).await
+ }
+ enterBarrier("blackhole")
+
+ runOn(third) {
+ cluster.shutdown()
+ }
+
+ runOn(first) {
+ expectNoMessage(2.second) // no ThisActorSystemQuarantinedEvent
+ testConductor.passThrough(first, fourth, ThrottlerTransportAdapter.Direction.Both).await
+ }
+
+ runOn(first) {
+ expectNoMessage(1.second) // no ThisActorSystemQuarantinedEvent
+ }
+ enterBarrier("wait")
+
+ runOn(first) {
+ val sel = system.actorSelection(RootActorPath(fourth) / "user")
+ (1 to 15).foreach { _ =>
+ sel.tell(Identify(None), ActorRef.noSender) // try to ping fourth
+ expectNoMessage(200.millis) // no ThisActorSystemQuarantinedEvent
+ }
+ }
+
+ enterBarrier("after-4")
}
}
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
index f0fc935036..19f6e3a025 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
@@ -560,8 +560,8 @@ private[remote] class Association(
// quarantine state change was performed
if (harmless) {
log.info(
- "Association to [{}] having UID [{}] has been stopped. All " +
- "messages to this UID will be delivered to dead letters. Reason: {}",
+ "Association to [{}] having UID [{}] has been stopped. Harmless quarantine. " +
+ "All messages to this UID will be delivered to dead letters. Reason: {}",
remoteAddress,
u,
reason)
@@ -585,6 +585,7 @@ private[remote] class Association(
send(ClearSystemMessageDelivery(current.incarnation), OptionVal.None, OptionVal.None)
if (!harmless) {
// try to tell the other system that we have quarantined it
+ log.info("Sending Quarantined to [{}]", peer)
sendControl(Quarantined(localAddress, peer))
}
setupStopQuarantinedTimer()
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/InboundQuarantineCheck.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/InboundQuarantineCheck.scala
index c6d2094149..b6e9002889 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/InboundQuarantineCheck.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/InboundQuarantineCheck.scala
@@ -61,10 +61,12 @@ private[remote] class InboundQuarantineCheck(inboundContext: InboundContext)
association.remoteAddress,
env.originUid)
// avoid starting outbound stream for heartbeats
- if (!env.message.isInstanceOf[Quarantined] && !isHeartbeat(env.message))
+ if (!env.message.isInstanceOf[Quarantined] && !isHeartbeat(env.message)) {
+ log.info("Sending Quarantined to [{}]", association.remoteAddress)
inboundContext.sendControl(
association.remoteAddress,
Quarantined(inboundContext.localAddress, UniqueAddress(association.remoteAddress, env.originUid)))
+ }
}
pull(in)
} else
diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/TestContext.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/TestContext.scala
index 0212214d83..e44960b55a 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/artery/TestContext.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/artery/TestContext.scala
@@ -103,7 +103,7 @@ private[remote] class TestOutboundContext(
}
override def quarantine(reason: String): Unit = synchronized {
- _associationState = _associationState.newQuarantined()
+ _associationState = _associationState.newQuarantined(harmless = false)
}
override def isOrdinaryMessageStreamActive(): Boolean = true
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org