pjfanning opened a new issue, #3552:
URL: https://github.com/apache/pekko/issues/3552
### Summary
On a single-node cluster with `remember-entities = on`,
`CoordinatedShutdown` takes ~9–10 seconds in the
`cluster-sharding-shutdown-region` phase when messages keep arriving for shards
while the region is shutting down. All shards hand off within milliseconds; the
rest of the time the `ShardRegion` sits with buffered messages, re-sending
`GetShardHome` every `retry-interval` (2s) to a coordinator that never answers,
until `GracefulShutdownTimeout` (phase timeout − 1s) fires and stops the region
anyway.
Nothing useful happens during that wait: with one member there is no other
region the shards could move to, and the buffered messages are dropped when the
timeout fires regardless.
Practical impact: the phase lands just under the 10s default
`cluster-sharding-shutdown-region` timeout, so test suites that start one
`ActorSystem` per class and use `TestKit.shutdownActorSystem` see a
multi-second stall per class, logged as a warning rather than failed.
### Reproduction
Pekko `main` at de6afb0459, JDK 17.0.19, macOS. Single-node cluster,
`PekkoSpec`, config:
```
pekko.actor.provider = "cluster"
pekko.remote.artery.canonical.port = 0
pekko.persistence.journal.plugin = "pekko.persistence.journal.inmem"
pekko.cluster.sharding.remember-entities = on
pekko.cluster.sharding.remember-entities-store = eventsourced
pekko.cluster.sharding.state-store-mode = ddata
```
1. `cluster.join(cluster.selfAddress)` and wait for `Up`.
2. `ClusterSharding(system).start(...)` with an `EchoActor` entity,
`extractShardId = msg.toString` so each entity is its own shard.
3. Send messages `1 to 14` and wait for each echo, so 14 shards are hosted.
4. Start a background thread that keeps sending `1 to 14` to the region
every 5ms.
5. `Await.result(CoordinatedShutdown(system).run(UnknownReason),
30.seconds)` and time it.
Observed: **10314ms**. Same run with no traffic during shutdown, or with
`remember-entities = off`, or with a single shard: ~20ms.
The shard count matters because the problem needs at least one shard to have
been deallocated while others are still alive: with one shard the region stops
on that shard's `Terminated` before anything can be buffered.
### Log timeline (Pekko, `verbose-debug-logging = on`)
```
23.550 type1: Starting graceful shutdown of region and all its shards
23.566 type1: Starting shutting down shards
[12,8,4,13,5,10,14,2,7,3,11,9,6,1] ...
23.683 type1: Shard [12] deallocation completed successfully.
23.759 type1: Shard [2] shutting down
...
23.774 type1: GetShardHome [12] request from [.../system/sharding/type1]
stashed, because waiting for initial state or update of state.
(133 stashed GetShardHome requests in total, from the region and
from pekko://local/ignore)
24.826 type1: Requesting shard home for [5] from coordinator ... [188]
buffered messages.
24.827 type1: Sending graceful shutdown to
List(.../type1Coordinator/singleton/coordinator)
26.845 type1: Requesting shard home for [5] from coordinator ... [520]
buffered messages.
28.864 type1: Requesting shard home for [5] from coordinator ... [852]
buffered messages.
30.886 type1: Requesting shard home for [5] from coordinator ... [1196]
buffered messages.
32.575 WARN type1: Graceful shutdown of shard region timed out, region will
be stopped. Remaining shards [], remaining buffered messages [20725].
32.576 type1: ShardRegion stopped
```
### Root cause
Two pieces of Pekko code interact:
**1. The region will not finish graceful shutdown while it holds buffered
messages** — `ShardRegion.tryCompleteGracefulShutdownIfInProgress`
(`cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala`):
```scala
if (gracefulShutdownInProgress && shards.isEmpty && shardBuffers.isEmpty) {
context.stop(self)
}
```
During graceful shutdown, once shard A has been handed off, a new message
for A goes through `deliverMessage` → `regionByShard.get(shardId)` is `None` →
the region sends `GetShardHome(A)` to the coordinator and buffers the message.
The buffer can only drain when a `ShardHome(A, otherRegion)` arrives.
**2. The coordinator silently ignores `GetShardHome` when there is no region
it could allocate to** — `ShardCoordinator.receive`:
```scala
case GetShardHome(shard) =>
if (!handleGetShardHome(shard)) {
val activeRegions = (state.regions -- gracefulShutdownInProgress) --
regionTerminationInProgress
if (activeRegions.nonEmpty) {
... allocate ...
}
// else: no reply, no log
}
```
On a single node the only region is in `gracefulShutdownInProgress`, so
`activeRegions` is empty and every request is dropped without a reply. The
region retries every `retry-interval`, gets nothing, and only exits when
`GracefulShutdownTimeout` stops it.
The "stashed, because waiting for initial state or update of state" lines
are a secondary effect: the ddata coordinator stashes `GetShardHome` while each
`ShardHomeDeallocated` write is in flight, and later replays them into the same
silent no-op. (Note also that the no-op branch never calls
`unstashOneGetShardHomeRequest()`, so replay only continues after the next
state update.)
`remember-entities` matters because each shard stop goes through the
remember-entities store, which widens the window between "shard A handed off"
and "last shard handed off" in which messages for A get buffered.
### Possible directions
- **Region side:** when graceful shutdown is in progress, all shards are
gone, and no other cluster member could host this region's shards (same data
centre, region role if configured), drop the buffered messages to dead letters
and stop the region immediately. Those messages are dropped by
`GracefulShutdownTimeout` today anyway; this only changes *when*. Multi-node
behaviour would be untouched: if another candidate member exists the region
keeps buffering so messages can be forwarded to the shard's new home. No
protocol change.
- **Coordinator side:** reply to `GetShardHome` when `activeRegions` is
empty so the region can react. This needs a new coordinator→region message,
hence a serializer change and rolling-upgrade considerations, so it is the
heavier option.
A regression test should assert the behaviour (buffered messages go to dead
letters and the region stops) rather than just the timing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
