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 7328e559e0 test: stabilize mixed protocol netty ssl cluster join
(#3203)
7328e559e0 is described below
commit 7328e559e0205e88881a11b479fb7e9b451dd8e2
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jun 28 01:56:30 2026 +0800
test: stabilize mixed protocol netty ssl cluster join (#3203)
Motivation:
MixedProtocolClusterSpec can fail intermittently in CI when classic netty
SSL gates a transiently failed association for the default 5 seconds while the
test is still using the default 6 second dilated awaitCond window.
Modification:
Use shorter test retry settings for the classic netty variants so seed
joins and remoting gate recovery can retry inside the assertion window. Keep
that retry tuning in a named classic remoting test config and collapse the
repeated join/setup/assert/shutdown flow into assertJoinSucceeds.
Result:
MixedProtocolClusterSpec no longer fails before the netty SSL recovery path
has a chance to retry and move the joining node to Up, and the test now states
each join scenario through its config pair instead of copy-pasted cluster
lifecycle code.
Tests:
- rtk sbt -Dpekko.test.timefactor=2 "cluster / Test / testOnly
org.apache.pekko.cluster.MixedProtocolClusterSpec -- -z \"allow a node using
the pekko protocol to join the cluster (netty ssl)\"" - succeeded, 1 test
- rtk sbt -Dpekko.test.timefactor=2 "cluster / Test / testOnly
org.apache.pekko.cluster.MixedProtocolClusterSpec" - succeeded, 9 tests
- rtk scalafmt --mode diff-ref=origin/main
- rtk scalafmt --list --mode diff-ref=origin/main
- rtk git diff --check
- rtk qodercli --help
- rtk bash -lc 'set -o pipefail; qodercli -p --output-format stream-json
--cwd "$PWD" --attachment /tmp/project-review.diff "Review the attached diff.
Focus on blocker/must-fix issues only. For every finding cite file:line
evidence. If none, say No must-fix findings and list residual risks." 2>&1 |
tee /tmp/project-qoder-review.log' - No must-fix findings
References:
Fixes #3202
---
.../pekko/cluster/MixedProtocolClusterSpec.scala | 163 +++++----------------
1 file changed, 34 insertions(+), 129 deletions(-)
diff --git
a/cluster/src/test/scala/org/apache/pekko/cluster/MixedProtocolClusterSpec.scala
b/cluster/src/test/scala/org/apache/pekko/cluster/MixedProtocolClusterSpec.scala
index 31da7e0f9f..8a69642f49 100644
---
a/cluster/src/test/scala/org/apache/pekko/cluster/MixedProtocolClusterSpec.scala
+++
b/cluster/src/test/scala/org/apache/pekko/cluster/MixedProtocolClusterSpec.scala
@@ -21,6 +21,7 @@ import org.apache.pekko
import pekko.testkit.{ LongRunningTest, PekkoSpec }
import com.typesafe.config.{ Config, ConfigFactory }
+import scala.concurrent.duration._
object MixedProtocolClusterSpec {
@@ -72,13 +73,20 @@ object MixedProtocolClusterSpec {
pekko.remote.protocol-name = "akka"
""").withFallback(baseConfig)
+ val configWithClassicRemotingTestRetries: Config =
+ ConfigFactory.parseString("""
+ # Classic remoting gates a failed association before retrying. Keep the
local
+ # test retry loop shorter than the join assertion window, especially for
netty ssl.
+ pekko.cluster.seed-node-timeout = 1 s
+ pekko.cluster.retry-unsuccessful-join-after = 1 s
+ pekko.remote.classic.retry-gate-closed-for = 1 s
+ """).withFallback(baseConfig)
+
val configWithNetty: Config =
ConfigFactory.parseString("""
pekko.remote.artery.enabled = false
- pekko.remote.classic {
- enabled-transports = ["pekko.remote.classic.netty.tcp"]
- }
- """).withFallback(baseConfig)
+ pekko.remote.classic.enabled-transports =
["pekko.remote.classic.netty.tcp"]
+ """).withFallback(configWithClassicRemotingTestRetries)
val configWithPekkoNetty: Config =
ConfigFactory.parseString("""
@@ -123,142 +131,51 @@ class MixedProtocolClusterSpec extends PekkoSpec with
ClusterTestKit {
import MixedProtocolClusterSpec._
- "A node using the akka protocol" must {
-
- "be allowed to join a cluster with a node using the pekko protocol (udp)"
taggedAs LongRunningTest in {
+ private def assertJoinSucceeds(firstNodeConfig: Config, joiningNodeConfig:
Config): Unit = {
+ val clusterTestUtil = new ClusterTestUtil(system.name)
+ try {
+ clusterTestUtil.newActorSystem(firstNodeConfig)
+ val joiningNode = clusterTestUtil.newActorSystem(joiningNodeConfig)
+ clusterTestUtil.formCluster()
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // start the first node with the "pekko" protocol
- clusterTestUtil.newActorSystem(configWithPekkoUdp)
+ awaitCond(clusterTestUtil.isMemberUp(joiningNode), max = 10.seconds,
message = "awaiting joining node to be 'Up'")
+ } finally {
+ clusterTestUtil.shutdownAll()
+ }
+ }
- // have a node using the "akka" protocol join
- val joiningNode = clusterTestUtil.newActorSystem(configWithAkkaUdp)
- clusterTestUtil.formCluster()
+ "A node using the akka protocol" must {
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ "be allowed to join a cluster with a node using the pekko protocol (udp)"
taggedAs LongRunningTest in {
+ assertJoinSucceeds(configWithPekkoUdp, configWithAkkaUdp)
}
"be allowed to join a cluster with a node using the pekko protocol (tcp)"
taggedAs LongRunningTest in {
-
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // start the first node with the "pekko" protocol
- clusterTestUtil.newActorSystem(configWithPekkoTcp)
-
- // have a node using the "akka" protocol join
- val joiningNode = clusterTestUtil.newActorSystem(configWithAkkaTcp)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(configWithPekkoTcp, configWithAkkaTcp)
}
"be allowed to join a cluster with a node using the pekko protocol
(netty)" taggedAs LongRunningTest in {
-
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // start the first node with the "pekko" protocol
- clusterTestUtil.newActorSystem(configWithPekkoNetty)
-
- // have a node using the "akka" protocol join
- val joiningNode = clusterTestUtil.newActorSystem(configWithAkkaNetty)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(configWithPekkoNetty, configWithAkkaNetty)
}
"be allowed to join a cluster with a node using the pekko protocol (netty
ssl)" taggedAs LongRunningTest in {
-
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // start the first node with the "pekko" protocol
- clusterTestUtil.newActorSystem(configWithPekkoNettySsl)
-
- // have a node using the "akka" protocol join
- val joiningNode =
clusterTestUtil.newActorSystem(configWithAkkaNettySsl)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(configWithPekkoNettySsl, configWithAkkaNettySsl)
}
"allow a node using the pekko protocol to join the cluster (udp)" taggedAs
LongRunningTest in {
-
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // create the first node with the "akka" protocol
- clusterTestUtil.newActorSystem(configWithAkkaUdp)
-
- // have a node using the "pekko" protocol join
- val joiningNode = clusterTestUtil.newActorSystem(configWithPekkoUdp)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(configWithAkkaUdp, configWithPekkoUdp)
}
"allow a node using the pekko protocol to join the cluster (tcp)" taggedAs
LongRunningTest in {
-
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // create the first node with the "akka" protocol
- clusterTestUtil.newActorSystem(configWithAkkaTcp)
-
- // have a node using the "pekko" protocol join
- val joiningNode = clusterTestUtil.newActorSystem(configWithPekkoTcp)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(configWithAkkaTcp, configWithPekkoTcp)
}
"allow a node using the pekko protocol to join the cluster (netty)"
taggedAs LongRunningTest in {
-
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // create the first node with the "akka" protocol
- clusterTestUtil.newActorSystem(configWithAkkaNetty)
-
- // have a node using the "pekko" protocol join
- val joiningNode = clusterTestUtil.newActorSystem(configWithPekkoNetty)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(configWithAkkaNetty, configWithPekkoNetty)
}
"allow a node using the pekko protocol to join the cluster (netty ssl)"
taggedAs LongRunningTest in {
-
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // create the first node with the "akka" protocol
- clusterTestUtil.newActorSystem(configWithAkkaNettySsl)
-
- // have a node using the "pekko" protocol join
- val joiningNode =
clusterTestUtil.newActorSystem(configWithPekkoNettySsl)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(configWithAkkaNettySsl, configWithPekkoNettySsl)
}
}
@@ -270,19 +187,7 @@ class MixedProtocolClusterSpec extends PekkoSpec with
ClusterTestKit {
val cfg = ConfigFactory.parseString("""
pekko.remote.accept-protocol-names = ["pekko"]""")
.withFallback(configWithPekkoNettySsl)
- val clusterTestUtil = new ClusterTestUtil(system.name)
- try {
- // create the first node with the "pekko" protocol
- clusterTestUtil.newActorSystem(cfg)
-
- // have a node using the "pekko" protocol join
- val joiningNode = clusterTestUtil.newActorSystem(cfg)
- clusterTestUtil.formCluster()
-
- awaitCond(clusterTestUtil.isMemberUp(joiningNode), message = "awaiting
joining node to be 'Up'")
- } finally {
- clusterTestUtil.shutdownAll()
- }
+ assertJoinSucceeds(cfg, cfg)
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
