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 3d68b6d67e fix: enforce cluster router limits for AdjustPoolSize
(#3364)
3d68b6d67e is described below
commit 3d68b6d67ec48b0c6b2e78f67841eeb39cf6efab
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sat Jul 18 21:22:09 2026 +0800
fix: enforce cluster router limits for AdjustPoolSize (#3364)
Motivation:
ClusterRouterPool inherited the generic AdjustPoolSize increase path, which
created local routees without applying cluster deployment constraints.
Modification:
Handle positive adjustments with the cluster deployment selector and cap the
number of additions at the requested delta. Add directional tests and
document
the constrained behavior.
Result:
AdjustPoolSize can no longer bypass eligible-node, total-instance, or
per-node-instance limits for cluster router pools.
Tests:
- ClusterRouterSupervisorSpec: passed (6 tests; 5 low-throughput repeats
also passed)
- cluster / Test / test: passed (370 tests)
- docs / paradox: passed
- +headerCheckAll, checkCodeStyle, and +mimaReportBinaryIssues: passed
- validatePullRequest: affected modules passed until docs / Pr-validation /
test;
that task had 634 passed and 1 existing IntegrationDocTest failure
because it
references pekko.dispatch.UnboundedMailbox
- scalafmt --list --mode diff-ref=origin/main and git diff --check: passed
- sortImports: failed in existing tooling with NoSuchMethodError for
scala.meta.package$.Input(); no unrelated output retained
References:
Fixes #3362
---
.../cluster/routing/ClusterRouterConfig.scala | 42 ++++----
.../routing/ClusterRouterSupervisorSpec.scala | 106 +++++++++++++++++++++
docs/src/main/paradox/cluster-routing.md | 4 +
3 files changed, 136 insertions(+), 16 deletions(-)
diff --git
a/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala
b/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala
index f499c8801a..80131e8125 100644
---
a/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala
+++
b/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala
@@ -29,6 +29,7 @@ import pekko.japi.Util.immutableSeq
import pekko.remote.RemoteScope
import pekko.routing.ActorRefRoutee
import pekko.routing.ActorSelectionRoutee
+import pekko.routing.AdjustPoolSize
import pekko.routing.Group
import pekko.routing.Pool
import pekko.routing.Resizer
@@ -406,28 +407,37 @@ private[pekko] class ClusterRouterPoolActor(
extends RouterPoolActor(supervisorStrategy)
with ClusterRouterActor {
- override def receive = clusterReceive.orElse(super.receive)
+ override def receive =
+ clusterReceive
+ .orElse({
+ case AdjustPoolSize(change) if change > 0 => addRoutees(change)
+ }: Actor.Receive)
+ .orElse(super.receive)
/**
* Adds routees based on totalInstances and maxInstancesPerNode settings
*/
- override def addRoutees(): Unit = {
- @tailrec
- def doAddRoutees(): Unit = selectDeploymentTarget match {
- case None => // done
- case Some(target) =>
- val routeeProps = cell.routeeProps
- val deploy =
- Deploy(config = ConfigFactory.empty(), routerConfig =
routeeProps.routerConfig, scope = RemoteScope(target))
- val routee = pool.newRoutee(routeeProps.withDeploy(deploy), context)
- // must register each one, since registered routees are used in
selectDeploymentTarget
- cell.addRoutee(routee)
+ override def addRoutees(): Unit = addRoutees(settings.totalInstances)
- // recursion until all created
- doAddRoutees()
- }
+ private def addRoutees(numberOfRoutees: Int): Unit = {
+ @tailrec
+ def doAddRoutees(remaining: Int): Unit =
+ if (remaining > 0)
+ selectDeploymentTarget match {
+ case None => // done
+ case Some(target) =>
+ val routeeProps = cell.routeeProps
+ val deploy =
+ Deploy(config = ConfigFactory.empty(), routerConfig =
routeeProps.routerConfig,
+ scope = RemoteScope(target))
+ val routee = pool.newRoutee(routeeProps.withDeploy(deploy),
context)
+ // must register each one, since registered routees are used in
selectDeploymentTarget
+ cell.addRoutee(routee)
+
+ doAddRoutees(remaining - 1)
+ }
- doAddRoutees()
+ doAddRoutees(numberOfRoutees)
}
def selectDeploymentTarget: Option[Address] = {
diff --git
a/cluster/src/test/scala/org/apache/pekko/cluster/routing/ClusterRouterSupervisorSpec.scala
b/cluster/src/test/scala/org/apache/pekko/cluster/routing/ClusterRouterSupervisorSpec.scala
index 52fcb0933e..c7c0fd4f37 100644
---
a/cluster/src/test/scala/org/apache/pekko/cluster/routing/ClusterRouterSupervisorSpec.scala
+++
b/cluster/src/test/scala/org/apache/pekko/cluster/routing/ClusterRouterSupervisorSpec.scala
@@ -16,7 +16,13 @@ package org.apache.pekko.cluster.routing
import org.apache.pekko
import pekko.actor._
import pekko.actor.OneForOneStrategy
+import pekko.cluster.Cluster
+import pekko.cluster.ClusterEvent.CurrentClusterState
+import pekko.cluster.MemberStatus
+import pekko.routing.AdjustPoolSize
+import pekko.routing.GetRoutees
import pekko.routing.RoundRobinPool
+import pekko.routing.Routees
import pekko.testkit._
object ClusterRouterSupervisorSpec {
@@ -40,6 +46,25 @@ class ClusterRouterSupervisorSpec extends PekkoSpec("""
import ClusterRouterSupervisorSpec._
+ private val cluster = Cluster(system)
+
+ override protected def atStartup(): Unit = {
+ cluster.join(cluster.selfAddress)
+ awaitAssert(cluster.selfMember.status should ===(MemberStatus.Up))
+ }
+
+ private def routeesAfterInitialClusterState(router: ActorRef): Routees = {
+ // Ensure preStart has sent the subscription before asking the cluster
event publisher for a barrier.
+ router.tell(GetRoutees, testActor)
+ expectMsgType[Routees]
+ cluster.sendCurrentClusterState(testActor)
+ expectMsgType[CurrentClusterState]
+
+ // The subscription snapshot was enqueued before the barrier reply, so
this query observes it.
+ router.tell(GetRoutees, testActor)
+ expectMsgType[Routees]
+ }
+
"Cluster aware routers" must {
"use provided supervisor strategy" in {
@@ -59,6 +84,87 @@ class ClusterRouterSupervisorSpec extends PekkoSpec("""
expectMsg("supervised")
}
+ "not create local routees through AdjustPoolSize when local routees are
disabled" in {
+ val router = system.actorOf(
+ ClusterRouterPool(
+ RoundRobinPool(nrOfInstances = 0),
+ ClusterRouterPoolSettings(totalInstances = 1, maxInstancesPerNode =
1, allowLocalRoutees = false))
+ .props(Props(classOf[KillableActor])))
+
+ router ! AdjustPoolSize(1)
+ router.tell(GetRoutees, testActor)
+
+ expectMsg(Routees(Vector.empty))
+ }
+
+ "not create local routees through AdjustPoolSize when local roles do not
match" in {
+ val router = system.actorOf(
+ ClusterRouterPool(
+ RoundRobinPool(nrOfInstances = 0),
+ ClusterRouterPoolSettings(
+ totalInstances = 1,
+ maxInstancesPerNode = 1,
+ allowLocalRoutees = true,
+ useRoles = Set("worker")))
+ .props(Props(classOf[KillableActor])))
+
+ router ! AdjustPoolSize(1)
+ router.tell(GetRoutees, testActor)
+
+ expectMsg(Routees(Vector.empty))
+ }
+
+ "not exceed the per-node limit through AdjustPoolSize" in {
+ val router = system.actorOf(
+ ClusterRouterPool(
+ RoundRobinPool(nrOfInstances = 0),
+ ClusterRouterPoolSettings(totalInstances = 2, maxInstancesPerNode =
1, allowLocalRoutees = true))
+ .props(Props(classOf[KillableActor])))
+
+ router ! AdjustPoolSize(1)
+ router ! AdjustPoolSize(1)
+ router.tell(GetRoutees, testActor)
+
+ expectMsgType[Routees].routees.size should ===(1)
+ }
+
+ "not exceed the total routee limit through AdjustPoolSize" in {
+ val router = system.actorOf(
+ ClusterRouterPool(
+ RoundRobinPool(nrOfInstances = 0),
+ ClusterRouterPoolSettings(totalInstances = 3, maxInstancesPerNode =
3, allowLocalRoutees = true))
+ .props(Props(classOf[KillableActor])))
+
+ routeesAfterInitialClusterState(router).routees.size should ===(3)
+
+ router ! AdjustPoolSize(-2)
+ router.tell(GetRoutees, testActor)
+ expectMsgType[Routees].routees.size should ===(1)
+
+ router ! AdjustPoolSize(5)
+ router.tell(GetRoutees, testActor)
+ expectMsgType[Routees].routees.size should ===(3)
+ }
+
+ "add only the requested number of routees through AdjustPoolSize" in {
+ val router = system.actorOf(
+ ClusterRouterPool(
+ RoundRobinPool(nrOfInstances = 0),
+ ClusterRouterPoolSettings(totalInstances = 3, maxInstancesPerNode =
3, allowLocalRoutees = true))
+ .props(Props(classOf[KillableActor])))
+
+ routeesAfterInitialClusterState(router).routees.size should ===(3)
+
+ router ! AdjustPoolSize(-2)
+ router.tell(GetRoutees, testActor)
+ expectMsgType[Routees].routees.size should ===(1)
+
+ router ! AdjustPoolSize(1)
+ router.tell(GetRoutees, testActor)
+
+ expectMsgType[Routees].routees.size should ===(2)
+ }
+
}
}
diff --git a/docs/src/main/paradox/cluster-routing.md
b/docs/src/main/paradox/cluster-routing.md
index 8d1abce7be..2e0eaba684 100644
--- a/docs/src/main/paradox/cluster-routing.md
+++ b/docs/src/main/paradox/cluster-routing.md
@@ -176,6 +176,10 @@ per node, `max-nr-of-instances-per-node`, will not be
exceeded. By default `max-
is set to a high value (10000) that will result in new routees added to the
router when nodes join the cluster.
Set it to a lower value if you want to limit total number of routees.
+Routees added with @apidoc[routing.AdjustPoolSize] are deployed only to
eligible nodes and are constrained by
+`max-total-nr-of-instances` and `max-nr-of-instances-per-node`. If fewer
eligible slots are available than requested,
+only the available number of routees are added.
+
The same type of router could also have been defined in code:
Scala
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
