pjfanning commented on code in PR #3701:
URL: https://github.com/apache/pekko/pull/3701#discussion_r4367756767
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala:
##########
@@ -548,6 +548,12 @@ private[pekko] class Shard(
receiveLeaseLost(ll)
case me: MemberEvent =>
receiveMemberEvent(me)
+ case HandOff(`shardId`) =>
Review Comment:
I think this can leak the lease. While in `awaitingLease`, the acquire
started by `tryGetLease` (line 578) is usually still pending. `handOff` stops
the shard, so `postStop` calls `lease.release()` for a lease the shard doesn't
hold yet. If the backend then grants the acquire, the `LeaseAcquireResult(true,
_)` goes to dead letters and nothing ever releases the lease. The shard's new
home on another node then sits in `awaitingLease` until the lease times out, or
for good if the implementation keeps renewing it.
A shard stopped by region shutdown during `awaitingLease` can already hit
this, but this change makes it routine, and in exactly the degraded-backend
case it targets. Could we release once the outstanding acquire completes, e.g.
track that an acquire is in flight and attach an `onComplete` that releases if
the result was `true`?
##########
cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala:
##########
@@ -162,6 +165,28 @@ class ShardWithLeaseSpec extends
PekkoSpec(ShardWithLeaseSpec.config) with WithL
sharding.tell(EntityEnvelope(1, "hello"), probe.ref)
probe.expectNoMessage(shortDuration)
}
+
+ "reply with ShardStopped on HandOff while awaiting lease" in new Setup {
+ val probe = TestProbe()
+ sharding.tell(EntityEnvelope(1, "hello"), probe.ref)
+ leaseFor("1") // the shard is created and the lease acquisition is
pending
+ probe.expectNoMessage(shortDuration)
+
+ val shard = identifyRef(s"/system/sharding/$typeName/1")
+ val coordinatorProbe = TestProbe()
+ watch(shard)
+ // the shard has not loaded any entities while awaiting the lease,
+ // so it must complete the hand off immediately instead of stashing it
+ shard.tell(ShardCoordinator.Internal.HandOff("1"), coordinatorProbe.ref)
+ coordinatorProbe.expectMsg(ShardCoordinator.Internal.ShardStopped("1"))
+ expectTerminated(shard)
Review Comment:
It would be good to also complete the pending acquire after the shard has
terminated, e.g. `leaseFor("1").initialPromise.complete(Success(true))`, and
then assert the lease ends up released. As written, the test passes even if the
late acquire leaves the lease held (see the comment on `Shard.scala:551`).
##########
cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala:
##########
@@ -162,6 +165,28 @@ class ShardWithLeaseSpec extends
PekkoSpec(ShardWithLeaseSpec.config) with WithL
sharding.tell(EntityEnvelope(1, "hello"), probe.ref)
probe.expectNoMessage(shortDuration)
}
+
+ "reply with ShardStopped on HandOff while awaiting lease" in new Setup {
+ val probe = TestProbe()
+ sharding.tell(EntityEnvelope(1, "hello"), probe.ref)
+ leaseFor("1") // the shard is created and the lease acquisition is
pending
+ probe.expectNoMessage(shortDuration)
+
+ val shard = identifyRef(s"/system/sharding/$typeName/1")
+ val coordinatorProbe = TestProbe()
+ watch(shard)
+ // the shard has not loaded any entities while awaiting the lease,
+ // so it must complete the hand off immediately instead of stashing it
+ shard.tell(ShardCoordinator.Internal.HandOff("1"), coordinatorProbe.ref)
Review Comment:
This sends `HandOff` straight to the shard and bypasses `ShardRegion`. In
production the region forwards `HandOff` and records the shard in `handingOff`.
Here the region sees an unexpected termination instead, so a regression in how
the region forwards `HandOff` to a starting shard would still pass. Could the
test drive the handoff through the region, e.g. send `HandOff` to the region
with the coordinator probe as sender?
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala:
##########
@@ -548,6 +548,12 @@ private[pekko] class Shard(
receiveLeaseLost(ll)
case me: MemberEvent =>
receiveMemberEvent(me)
+ case HandOff(`shardId`) =>
+ // no entities have been started while awaiting the lease, so hand off
completes immediately;
+ // stashing it would prevent the rebalance from completing while the
lease is unobtainable
+ handOff(sender())
Review Comment:
Anything already stashed in `awaitingLease` is lost when the shard stops
here: `postStop` returns the stash to the mailbox and it goes to dead letters.
That includes `GetCurrentShardState` / `GetShardStats` queries forwarded by the
region, and `StartEntity` from `RememberEntityStarterManager`, so those callers
get no reply and just time out. Is that acceptable, or should we reply to or
forward those before handing off?
Also, this comment assumes `entities` is empty, but `handOff` still runs its
full active-entities / `HandOffStopper` branching. That is correct today only
because nothing registers entities before the lease is acquired, and nothing
asserts it.
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/Shard.scala:
##########
@@ -548,6 +548,12 @@ private[pekko] class Shard(
receiveLeaseLost(ll)
case me: MemberEvent =>
receiveMemberEvent(me)
+ case HandOff(`shardId`) =>
+ // no entities have been started while awaiting the lease, so hand off
completes immediately;
+ // stashing it would prevent the rebalance from completing while the
lease is unobtainable
+ handOff(sender())
+ case HandOff(shard) =>
+ log.warning("{}: Shard [{}] can not hand off for another Shard [{}]",
typeName, shardId, shard)
Review Comment:
nit: these two cases duplicate `receiveCoordinatorMessage` (line 875). `case
msg: CoordinatorMessage => receiveCoordinatorMessage(msg)` would reuse it, so
the handoff routing and the warning text live in one place.
--
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]
