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 c37d875caf fix: stop entities stuck in passivation after a timeout
(#3573)
c37d875caf is described below
commit c37d875caf761bf341619d8b7aa8fdfe9c1a7d47
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 9 08:38:13 2026 +0100
fix: stop entities stuck in passivation after a timeout (#3573)
Motivation:
When a sharded entity is passivated (explicitly with `Passivate` or by an
automatic passivation strategy), the `Shard` sends it the stop message and
keeps it in the `Passivating` state until it terminates, buffering all messages
for it in the meantime. An entity that never stops itself (for example because
it does not handle the stop message) stays passivating forever and messages for
it are buffered indefinitely.
Modification:
- New setting `pekko.cluster.sharding.passivation.stop-timeout` (default
`60s`, `off`/`none` disables it). This mirrors how `HandOffStopper` forcefully
stops entities after `handoff-timeout`.
- `Shard` starts a per-entity timer when it sends the stop message for
passivation. If the entity is still passivating when the timer fires, the shard
logs a warning and stops the entity. The existing termination path then
delivers any buffered messages to a new incarnation. The timer is cancelled
when the entity terminates.
- The setting is read from the system config the same way as
`verbose-debug-logging`, so there is no public API or binary change.
- Documented in the typed Cluster Sharding passivation section.
Result:
Entities that do not stop after passivation no longer get stuck with their
messages buffered forever.
Tests:
- `sbt "cluster-sharding/testOnly *PassivationStopTimeout*"`: passed.
`PassivationStopTimeoutSpec` fails without the `Shard` change (no restart,
buffered message not delivered).
- `sbt "cluster-sharding/Test/testOnly
org.apache.pekko.cluster.sharding.*"`: 256 tests passed
- `sbt cluster-sharding/scalafmtCheckAll cluster-sharding/headerCheckAll
docs/paradox`: passed
- `git diff --check`: clean
- MiMa not run: no public API or binary shape changes (only private members
of the internal `Shard` actor)
- Multi-JVM sharding tests not run locally
References:
None - independent fix for entities getting stuck in passivation
---
cluster-sharding/src/main/resources/reference.conf | 8 ++
.../org/apache/pekko/cluster/sharding/Shard.scala | 33 +++++
.../passivation/PassivationStopTimeoutSpec.scala | 152 +++++++++++++++++++++
docs/src/main/paradox/typed/cluster-sharding.md | 5 +
4 files changed, 198 insertions(+)
diff --git a/cluster-sharding/src/main/resources/reference.conf
b/cluster-sharding/src/main/resources/reference.conf
index f49250ee39..1127d6962c 100644
--- a/cluster-sharding/src/main/resources/reference.conf
+++ b/cluster-sharding/src/main/resources/reference.conf
@@ -43,6 +43,14 @@ pekko.cluster.sharding {
# may change after additional testing and feedback.
strategy = "default-idle-strategy"
+ # When an entity is passivated, either explicitly with `Passivate` or by
the automatic passivation
+ # strategy, it is sent its stop message and it is expected to stop itself.
Messages for the entity
+ # are buffered until it has stopped. If the entity has not stopped within
this timeout it is
+ # stopped forcefully, to avoid it being stuck in the passivating state
forever.
+ # Applies regardless of the automatic passivation strategy, including when
`remember-entities` is enabled.
+ # Set to "off" or "none" to wait indefinitely for the entity to stop.
+ stop-timeout = 60s
+
# Default passivation strategy without active entity limit; time out idle
entities after 2 minutes.
default-idle-strategy {
idle-entity.timeout = 120s
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala
index b45ba8e870..57785da53b 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala
@@ -49,6 +49,7 @@ import pekko.coordination.lease.scaladsl.LeaseProvider
import pekko.event.LoggingAdapter
import pekko.pattern.pipe
import pekko.util.Clock
+import pekko.util.Helpers.toRootLowerCase
import pekko.util.MessageBufferMap
import pekko.util.OptionVal
import pekko.util.PrettyDuration._
@@ -124,6 +125,8 @@ private[pekko] object Shard {
private final case class EntityTerminated(ref: ActorRef)
+ private final case class PassivationStopTimeout(entityId: EntityId) extends
NoSerializationVerificationNeeded
+
@nowarn("msg=never used")
private final case class RememberedEntityIds(ids: Set[EntityId])
private final case class RememberEntityStoreCrashed(store: ActorRef)
@@ -467,6 +470,15 @@ private[pekko] class Shard(
new Entities(log, settings.rememberEntities, verboseDebug,
failOnInvalidStateTransition)
}
+ private val passivationStopTimeout: Option[FiniteDuration] = {
+ val config = context.system.settings.config
+ val path = "pekko.cluster.sharding.passivation.stop-timeout"
+ toRootLowerCase(config.getString(path)) match {
+ case "off" | "none" => None
+ case _ => Some(config.getDuration(path,
MILLISECONDS).millis)
+ }
+ }
+
// Messages are buffered while an entity is passivating or waiting for a
response
// for that entity from the remember store
private val messageBuffers = new MessageBufferMap[EntityId]
@@ -633,6 +645,7 @@ private[pekko] class Shard(
case msg: ShardRegion.ShardsUpdated => shardsUpdated(msg)
case Passivate(stopMessage) => passivate(sender(),
stopMessage)
case PassivateIntervalTick =>
passivateEntitiesAfterInterval()
+ case PassivationStopTimeout(entityId) =>
passivationStopTimedOut(entityId)
case msg: ShardQuery => receiveShardQuery(msg)
case msg: LeaseLost => receiveLeaseLost(msg)
case msg: RememberEntityStoreCrashed =>
rememberEntityStoreCrashed(msg)
@@ -710,6 +723,7 @@ private[pekko] class Shard(
typeName,
entities.entityId(sender()).getOrElse(s"Unknown actor ${sender()}"))
passivate(sender(), stopMessage)
+ case PassivationStopTimeout(entityId) =>
passivationStopTimedOut(entityId)
case msg: ShardQuery => receiveShardQuery(msg)
case PassivateIntervalTick => stash()
case msg: RememberEntityStoreCrashed =>
rememberEntityStoreCrashed(msg)
@@ -932,6 +946,7 @@ private[pekko] class Shard(
}
case Passivating(_) =>
+ timers.cancel(PassivationStopTimeout(entityId))
if (rememberEntitiesStore.isDefined) {
if (entities.pendingRememberedEntitiesExist()) {
// will go in next batch update
@@ -991,12 +1006,30 @@ private[pekko] class Shard(
entities.entityPassivating(id)
entity ! stopMessage
flightRecorder.entityPassivate(id)
+ passivationStopTimeout.foreach { timeout =>
+ val msg = PassivationStopTimeout(id)
+ timers.startSingleTimer(msg, msg, timeout)
+ }
}
case _ =>
log.debug("{}: Unknown entity passivating [{}]. Not sending
stopMessage back to entity", typeName, entity)
}
}
+ private def passivationStopTimedOut(entityId: EntityId): Unit = {
+ entities.entityState(entityId) match {
+ case Passivating(ref) =>
+ log.warning(
+ "{}: Entity [{}] did not stop within [{}] after being sent its stop
message for passivation, stopping it",
+ typeName,
+ entityId,
+ passivationStopTimeout.map(_.pretty).getOrElse(""))
+ context.stop(ref)
+ case _ =>
+ // entity already stopped, or has been restarted since passivation
started
+ }
+ }
+
private def activeEntityLimitUpdated(updated:
ShardRegion.SetActiveEntityLimit): Unit = {
val entitiesToPassivate =
passivationStrategy.limitUpdated(updated.perRegionLimit)
passivateEntities(entitiesToPassivate)
diff --git
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/passivation/PassivationStopTimeoutSpec.scala
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/passivation/PassivationStopTimeoutSpec.scala
new file mode 100644
index 0000000000..5ac28c09eb
--- /dev/null
+++
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/passivation/PassivationStopTimeoutSpec.scala
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.cluster.sharding.passivation
+
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.Actor
+import pekko.actor.ActorRef
+import pekko.actor.Props
+import pekko.cluster.Cluster
+import pekko.cluster.sharding.ClusterSharding
+import pekko.cluster.sharding.ClusterShardingSettings
+import pekko.cluster.sharding.ShardRegion
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestProbe
+import pekko.testkit.WithLogCapturing
+
+import com.typesafe.config.Config
+import com.typesafe.config.ConfigFactory
+
+object PassivationStopTimeoutSpec {
+
+ val config: Config = ConfigFactory.parseString("""
+ pekko.loglevel = DEBUG
+ pekko.loggers = ["org.apache.pekko.testkit.SilenceAllTestEventListener"]
+ pekko.actor.provider = "cluster"
+ pekko.remote.classic.netty.tcp.port = 0
+ pekko.remote.artery.canonical.port = 0
+ pekko.cluster.sharding.verbose-debug-logging = on
+ pekko.cluster.sharding.fail-on-invalid-entity-state-transition = on
+ pekko.cluster.sharding.passivation.strategy = none
+ """)
+
+ val stopTimeoutConfig: Config =
+ ConfigFactory.parseString("pekko.cluster.sharding.passivation.stop-timeout
= 1s").withFallback(config)
+
+ val noStopTimeoutConfig: Config =
+ ConfigFactory.parseString("pekko.cluster.sharding.passivation.stop-timeout
= off").withFallback(config)
+
+ object Entity {
+ // the entity never stops itself when receiving this stop message
+ case object IgnoredStop
+ case object ManuallyPassivate
+ final case class Envelope(id: String, message: Any)
+ final case class Started(id: String)
+ final case class Received(id: String, message: Any)
+
+ def props(probe: ActorRef): Props = Props(new Entity(probe))
+ }
+
+ class Entity(probe: ActorRef) extends Actor {
+ private def id = self.path.name
+
+ override def preStart(): Unit = probe ! Entity.Started(id)
+
+ def receive: Receive = {
+ case Entity.ManuallyPassivate =>
+ probe ! Entity.Received(id, Entity.ManuallyPassivate)
+ context.parent ! ShardRegion.Passivate(Entity.IgnoredStop)
+ case msg => probe ! Entity.Received(id, msg)
+ }
+ }
+
+ val extractEntityId: ShardRegion.ExtractEntityId = {
+ case Entity.Envelope(id, message) => (id, message)
+ }
+
+ val extractShardId: ShardRegion.ExtractShardId = {
+ case Entity.Envelope(id, _) => id
+ case _ => throw new IllegalArgumentException
+ }
+}
+
+abstract class AbstractPassivationStopTimeoutSpec(config: Config) extends
PekkoSpec(config) with WithLogCapturing {
+ import PassivationStopTimeoutSpec._
+
+ val probe: TestProbe = TestProbe()
+
+ def start(): ActorRef = {
+ // single node cluster
+ Cluster(system).join(Cluster(system).selfAddress)
+ val settings = ClusterShardingSettings(system)
+ ClusterSharding(system).start(
+ "myType",
+ Entity.props(probe.ref),
+ settings,
+ extractEntityId,
+ extractShardId,
+ ClusterSharding(system).defaultShardAllocationStrategy(settings),
+ Entity.IgnoredStop)
+ }
+
+ def passivateEntityThatDoesNotStop(region: ActorRef): Unit = {
+ region ! Entity.Envelope("1", "A")
+ probe.expectMsg(Entity.Started("1"))
+ probe.expectMsg(Entity.Received("1", "A"))
+
+ region ! Entity.Envelope("1", Entity.ManuallyPassivate)
+ probe.expectMsg(Entity.Received("1", Entity.ManuallyPassivate))
+ probe.expectMsg(Entity.Received("1", Entity.IgnoredStop))
+
+ // buffered by the shard while the entity is passivating
+ region ! Entity.Envelope("1", "B")
+ }
+}
+
+class PassivationStopTimeoutSpec
+ extends
AbstractPassivationStopTimeoutSpec(PassivationStopTimeoutSpec.stopTimeoutConfig)
{
+ import PassivationStopTimeoutSpec._
+
+ "Passivation of an entity that does not stop" must {
+ "stop the entity after the stop timeout and deliver buffered messages to a
new incarnation" in {
+ val region = start()
+ passivateEntityThatDoesNotStop(region)
+ probe.expectNoMessage(500.millis)
+ probe.expectMsg(5.seconds, Entity.Started("1"))
+ probe.expectMsg(Entity.Received("1", "B"))
+
+ // the new incarnation is active and receives messages directly
+ region ! Entity.Envelope("1", "C")
+ probe.expectMsg(Entity.Received("1", "C"))
+ }
+ }
+}
+
+class PassivationStopTimeoutOffSpec
+ extends
AbstractPassivationStopTimeoutSpec(PassivationStopTimeoutSpec.noStopTimeoutConfig)
{
+
+ "Passivation of an entity that does not stop" must {
+ "keep the entity passivating when the stop timeout is off" in {
+ val region = start()
+ passivateEntityThatDoesNotStop(region)
+ probe.expectNoMessage(2.seconds)
+ }
+ }
+}
diff --git a/docs/src/main/paradox/typed/cluster-sharding.md
b/docs/src/main/paradox/typed/cluster-sharding.md
index cdbbb8e3a6..9ed97a3fc5 100644
--- a/docs/src/main/paradox/typed/cluster-sharding.md
+++ b/docs/src/main/paradox/typed/cluster-sharding.md
@@ -319,6 +319,11 @@ the entity when it's supposed to stop itself due to
rebalance or passivation. If
it will be stopped automatically without receiving a specific message. It can
be useful to define a custom stop
message if the entity needs to perform some asynchronous cleanup or
interactions before stopping.
+If a passivating entity has not stopped within
`pekko.cluster.sharding.passivation.stop-timeout` (default 60 seconds)
+after it was sent the stop message, it is stopped forcefully. This prevents an
entity that never stops itself, for
+example because it does not handle the stop message, from being stuck in
passivation with messages for it buffered
+indefinitely. Set the timeout to `off` to wait indefinitely for the entity to
stop itself.
+
The stop message is only sent locally, from the shard to the entity so does
not require an entity id to end up in the right actor. When using a custom
@apidoc[typed.ShardingMessageExtractor] without envelopes, the extractor will
still have to handle the stop message type to please the compiler, even though
it will never actually be passed to the extractor.
## Automatic Passivation
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
