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 9ba70fdef0 fix: report the effective timeout in shard hand off stop
warnings (#3644)
9ba70fdef0 is described below
commit 9ba70fdef0dc21840fe20317db89e2b53aea3c03
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 13:49:49 2026 +0100
fix: report the effective timeout in shard hand off stop warnings (#3644)
* fix: report the effective timeout in shard hand off stop warnings
Motivation:
The shard computed a reduced entity stop timeout (hand-off-timeout minus
5 seconds, at least 1 second) before passing it to HandOffStopper, which
then logged that reduced value as if it was the configured hand off
timeout. The early warning also claimed it would wait the whole timeout
again, although only the remainder is left, and the warning timer was not
cancelled when the stop timeout fired first.
Modification:
- Pass the configured hand-off-timeout to HandOffStopper and derive the
entity stop timeout inside it.
- Log the remaining wait time in the early warning.
- Cancel the warning timer when the stop timeout fires.
- Widen the expectations in ClusterShardingInternalsSpec since the
HandOffStopper entity stop timeout is now at least 1 second.
Result:
Hand off warnings report accurate timeouts.
Tests:
- Not run locally - left to CI; scalafmt applied
References:
None - log message improvement for shard hand off
* test: add directional HandOffStopper regression tests
Motivation:
Review of #3644 noted that widening the ClusterShardingInternalsSpec waits
also
accepts the old behaviour, and asked for a directional regression of the
remaining-time warning and the cancellation of the pending warning.
Modification:
- Add HandOffStopperSpec, which drives the HandOffStopper timers with
ExplicitlyTriggeredScheduler and checks the published warnings:
- a 20s hand-off-timeout warns after 5s with "Waiting additional
[10 seconds]" and stops the entities after 15s;
- after a 1s entity stop timeout, an entity that is still stopping does
not
lead to the 5s warning.
- Update the Lightbend notices of the files changed by the port to the
Akka 2.9.0 source notice (2009-2023).
Result:
The tests fail on the previous HandOffStopper (wrong remaining time and stop
delay) and when the warning cancellation is removed (it then logs "Waiting
additional [-4 seconds]").
Tests:
sbt "cluster-sharding/testOnly
org.apache.pekko.cluster.sharding.HandOffStopperSpec
org.apache.pekko.cluster.sharding.ClusterShardingInternalsSpec"
(4 passed; HandOffStopperSpec fails 2/2 against the old HandOffStopper and
1/2 with only timers.cancel removed)
sbt "cluster-sharding/headerCheckAll"
References:
https://github.com/apache/pekko/pull/3644
---
.../org/apache/pekko/cluster/sharding/Shard.scala | 14 ++-
.../pekko/cluster/sharding/ShardRegion.scala | 10 +-
.../sharding/ClusterShardingInternalsSpec.scala | 7 +-
.../cluster/sharding/HandOffStopperSpec.scala | 111 +++++++++++++++++++++
4 files changed, 131 insertions(+), 11 deletions(-)
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 0ade397b26..27ed10a2de 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
@@ -899,7 +899,6 @@ private[pekko] class Shard(
replyTo ! ShardStopped(shardId)
context.stop(self)
} else if (activeEntities.nonEmpty && !preparingForShutdown) {
- val entityHandOffTimeout = (settings.tuningParameters.handOffTimeout -
5.seconds).max(1.seconds)
log.debug(
"{}: Starting HandOffStopper for shard [{}] to terminate [{}]
entities.",
typeName,
@@ -907,9 +906,16 @@ private[pekko] class Shard(
activeEntities.size)
activeEntities.foreach(context.unwatch)
handOffStopper = Some(
- context.watch(context.actorOf(
- HandOffStopper.props(typeName, shardId, replyTo, activeEntities,
handOffStopMessage, entityHandOffTimeout),
- "HandOffStopper")))
+ context.watch(
+ context.actorOf(
+ HandOffStopper.props(
+ typeName,
+ shardId,
+ replyTo,
+ activeEntities,
+ handOffStopMessage,
+ settings.tuningParameters.handOffTimeout),
+ "HandOffStopper")))
// During hand off we only care about watching for termination of the
hand off stopper
context.become {
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
index 393de2195c..c2b837e681 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.cluster.sharding
@@ -534,8 +534,9 @@ object ShardRegion {
import HandOffStopper._
import ShardCoordinator.Internal.ShardStopped
+ val entityHandOffTimeout = (handoffTimeout - 5.seconds).max(1.seconds)
timers.startSingleTimer(StopTimeoutWarning, StopTimeoutWarning,
StopTimeoutWarningAfter)
- timers.startSingleTimer(StopTimeout, StopTimeout, handoffTimeout)
+ timers.startSingleTimer(StopTimeout, StopTimeout, entityHandOffTimeout)
entities.foreach { a =>
context.watch(a)
@@ -562,7 +563,7 @@ object ShardRegion {
if
(CoordinatedShutdown(context.system).getShutdownReason().isPresent)
"" // the region will be shutdown earlier so would be confusing to
say more
else
- s"Waiting additional [${handoffTimeout.toCoarsest}] before
stopping the remaining entities.")
+ s"Waiting additional [${(entityHandOffTimeout -
StopTimeoutWarningAfter).toCoarsest}] before stopping the remaining entities.")
case StopTimeout =>
log.warning(
@@ -570,8 +571,9 @@ object ShardRegion {
"stopping the remaining [{}] entities.",
stopMessage.getClass.getName,
shard,
- handoffTimeout.toCoarsest,
+ entityHandOffTimeout.toCoarsest,
remaining.size)
+ timers.cancel(StopTimeoutWarning)
remaining.foreach { ref =>
context.stop(ref)
diff --git
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ClusterShardingInternalsSpec.scala
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ClusterShardingInternalsSpec.scala
index 7a56163758..7daeb7afae 100644
---
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ClusterShardingInternalsSpec.scala
+++
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ClusterShardingInternalsSpec.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.cluster.sharding
@@ -98,10 +98,11 @@ class ClusterShardingInternalsSpec extends PekkoSpec("""
val handOffStopper = system.actorOf(
Props(new HandOffStopper(typeName, shard, probe.ref,
Set(emptyHandlerActor), HandOffStopMessage, 10.millis)))
+ // the entity stop timeout is at least 1 second
watch(emptyHandlerActor)
- expectTerminated(emptyHandlerActor, 1.seconds)
+ expectTerminated(emptyHandlerActor, 3.seconds)
- probe.expectMsg(1.seconds, ShardStopped(shard))
+ probe.expectMsg(3.seconds, ShardStopped(shard))
probe.lastSender shouldEqual handOffStopper
watch(handOffStopper)
diff --git
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/HandOffStopperSpec.scala
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/HandOffStopperSpec.scala
new file mode 100644
index 0000000000..9a05d5fc1d
--- /dev/null
+++
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/HandOffStopperSpec.scala
@@ -0,0 +1,111 @@
+/*
+ * 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
+
+import java.util.concurrent.{ CountDownLatch, TimeUnit }
+
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.{ Actor, Props }
+import pekko.cluster.sharding.ShardCoordinator.Internal.ShardStopped
+import pekko.cluster.sharding.ShardRegion.HandOffStopper
+import pekko.event.Logging
+import pekko.testkit.{ ExplicitlyTriggeredScheduler, PekkoSpec, TestProbe }
+
+object HandOffStopperSpec {
+ case object StopMessage
+
+ // does not handle the hand off stop message
+ class IgnoringEntity extends Actor {
+ override def receive: Receive = {
+ case _ =>
+ }
+ }
+
+ // blocks in postStop, so the HandOffStopper keeps running after it has
stopped this entity
+ class SlowStoppingEntity(release: CountDownLatch) extends Actor {
+ override def receive: Receive = {
+ case _ =>
+ }
+
+ override def postStop(): Unit = release.await(10, TimeUnit.SECONDS)
+ }
+}
+
+class HandOffStopperSpec extends PekkoSpec("""
+ pekko.scheduler.implementation =
"org.apache.pekko.testkit.ExplicitlyTriggeredScheduler"
+ """) {
+ import HandOffStopperSpec._
+
+ private def scheduler =
system.scheduler.asInstanceOf[ExplicitlyTriggeredScheduler]
+
+ private def warningsProbe(): TestProbe = {
+ val probe = TestProbe()
+ system.eventStream.subscribe(probe.ref, classOf[Logging.Warning])
+ probe
+ }
+
+ private def expectWarning(warnings: TestProbe, texts: String*): Unit =
+ warnings.fishForMessage(hint = texts.mkString(" ... ")) {
+ case w: Logging.Warning => texts.forall(w.message.toString.contains)
+ }
+
+ "HandOffStopper" must {
+
+ "report the remaining wait and stop the entities after the entity stop
timeout" in {
+ val warnings = warningsProbe()
+ val replyTo = TestProbe()
+ val entity = system.actorOf(Props[IgnoringEntity]())
+ // a hand-off-timeout of 20s gives an entity stop timeout of 15s
+ system.actorOf(HandOffStopper.props("typeName", "1", replyTo.ref,
Set(entity), StopMessage, 20.seconds))
+
+ scheduler.timePasses(5.seconds)
+ expectWarning(
+ warnings,
+ "not stopped after [5 seconds]",
+ "Waiting additional [10 seconds] before stopping the remaining
entities")
+
+ scheduler.timePasses(10.seconds)
+ expectWarning(warnings, "after [15 seconds], stopping the remaining [1]
entities")
+ replyTo.expectMsg(ShardStopped("1"))
+ }
+
+ "not warn after the entity stop timeout has fired" in {
+ val warnings = warningsProbe()
+ val replyTo = TestProbe()
+ val release = new CountDownLatch(1)
+ val entity = system.actorOf(Props(new SlowStoppingEntity(release)))
+ // a hand-off-timeout of 6s gives an entity stop timeout of 1s, which
fires before the 5s warning
+ system.actorOf(HandOffStopper.props("typeName", "2", replyTo.ref,
Set(entity), StopMessage, 6.seconds))
+
+ scheduler.timePasses(1.second)
+ expectWarning(warnings, "after [1 second], stopping the remaining [1]
entities")
+
+ // the entity is still stopping, so the HandOffStopper is still running
when the warning would be due
+ scheduler.timePasses(4.seconds)
+ release.countDown()
+ replyTo.expectMsg(ShardStopped("2"))
+
+ // a warning is published before the HandOffStopper handles the entity
termination and replies
+ warnings.receiveWhile(100.millis) {
+ case w: Logging.Warning => w.message.toString
+ }.filter(_.contains("not stopped after")) shouldBe empty
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
