pjfanning commented on code in PR #3709:
URL: https://github.com/apache/pekko/pull/3709#discussion_r4368073088
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala:
##########
@@ -966,8 +967,23 @@ abstract class ShardCoordinator(
}: Receive).orElse[Any, Unit](receiveTerminated)
private def terminate(): Unit = {
- if (aliveRegions.exists(_.path.address.hasLocalScope) ||
gracefulShutdownInProgress.exists(
- _.path.address.hasLocalScope)) {
+ val selfIsLeavingCluster = {
Review Comment:
`cluster.selfMember` comes from `ClusterReadView`, which is updated by its
own subscriber actor, separately from `ClusterSingletonManager`'s subscription.
In a real leave or self-down, the singleton manager can act on
`Leaving`/`Exiting`/`MemberDowned` (or a `HandOverToMe` from the new oldest)
and send `Terminate` before the read view has applied that status. If
`CoordinatedShutdown` hasn't started yet, `selfIsLeavingCluster` is `false` and
the coordinator stops without draining the local region. The hand-off
optimisation is then silently skipped, and shards on the leaving node can be
reallocated while their entities are still alive there.
More generally, the root issue is that the singleton manager sends the same
`terminationMessage` for lease loss and for hand-over, so the coordinator has
to guess why from membership. Telling it the actual reason would avoid the
race, e.g. a distinct termination message or flag for lease-loss stops from
`ClusterSingletonManager`. That touches cluster-tools, so it could be a
follow-up, but it seems worth considering before relying on this check.
##########
docs/src/main/paradox/typed/cluster-sharding.md:
##########
@@ -826,6 +826,12 @@ the owner is set to the
`Cluster(system).selfAddress.hostPort`.
If a shard can't acquire a lease it will remain uninitialized so messages for
entities it owns will
be buffered in the `ShardRegion`. If the lease is lost after initialization
the Shard will be terminated.
+A lease can also be used for the coordinator singleton by setting it in the
+`pekko.cluster.sharding.coordinator-singleton` properties. If the coordinator
singleton loses its
Review Comment:
nit: this applies equally to classic sharding (same `ShardCoordinator`), but
the classic `docs/src/main/paradox/cluster-sharding.md` doesn't mention it.
Worth adding the same note there, or linking to it.
##########
cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ClusterShardingCoordinatorLeaseLossSpec.scala:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.net.URLEncoder
+import java.nio.charset.StandardCharsets
+
+import scala.concurrent.duration._
+import scala.util.Success
+
+import org.apache.pekko
+import pekko.cluster.Cluster
+import pekko.cluster.MemberStatus
+import pekko.coordination.lease.TestLease
+import pekko.coordination.lease.TestLeaseExt
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestActors
+import pekko.testkit.TestException
+
+import com.typesafe.config.ConfigFactory
+
+object ClusterShardingCoordinatorLeaseLossSpec {
+ val config = ConfigFactory.parseString("""
+ pekko.loglevel = INFO
+ pekko.actor.provider = cluster
+ pekko.remote.classic.netty.tcp.port = 0
+ pekko.remote.artery.canonical.port = 0
+ pekko.cluster.sharding.coordinator-singleton {
+ use-lease = "test-lease"
+ lease-retry-interval = 200ms
+ }
+ """).withFallback(TestLease.config)
+}
+
+class ClusterShardingCoordinatorLeaseLossSpec
+ extends PekkoSpec(ClusterShardingCoordinatorLeaseLossSpec.config)
+ with ImplicitSender {
+
+ private val cluster = Cluster(system)
+
+ override protected def atStartup(): Unit = {
+ cluster.join(cluster.selfAddress)
+ awaitAssert(cluster.selfMember.status shouldEqual MemberStatus.Up)
+ }
+
+ private val extractEntityId: ShardRegion.ExtractEntityId = {
+ case i: Int => (i.toString, i)
+ }
+
+ private val extractShardId: ShardRegion.ExtractShardId = {
+ case i: Int => (i % 10).toString
+ case _ => "0"
+ }
+
+ "Cluster sharding with a coordinator singleton lease" should {
+
+ "keep the local shard region alive when the coordinator lease is lost" in {
+ val typeName = "coordinator-lease-loss"
+ val region = ClusterSharding(system).start(
+ typeName,
+ TestActors.echoActorProps,
+ ClusterShardingSettings(system),
+ extractEntityId,
+ extractShardId)
+
+ // the coordinator singleton acquires a lease before starting
+ val encName = URLEncoder.encode(typeName, StandardCharsets.UTF_8.name())
+ val leaseName =
s"${system.name}-singleton-pekko://${system.name}/system/sharding/${encName}Coordinator"
+ val coordinatorLease = awaitAssert {
+ TestLeaseExt(system).getTestLease(leaseName)
+ }
+ coordinatorLease.setNextCheckLeaseResult(true)
+ // the coordinator singleton acquires the lease before starting
+
coordinatorLease.probe.expectMsg(TestLease.AcquireReq(cluster.selfAddress.hostPort))
+ coordinatorLease.initialPromise.complete(Success(true))
+
+ // entity is reachable
+ region ! 1
+ expectMsg(10.seconds, 1)
+
+ // a transient lease backend problem: the singleton manager terminates
the
+ // coordinator and tries to re-acquire the lease while this node stays Up
+ coordinatorLease.getCurrentCallback()(Some(TestException("blip")))
+
+ // wait for the re-acquisition; without the fix the local region has been
+ // gracefully shut down by then and can never serve entities again
+ val owner = cluster.selfAddress.hostPort
+ coordinatorLease.probe.fishForMessage(10.seconds, "lease re-acquired
after loss") {
+ case TestLease.AcquireReq(`owner`) => true
+ case _ => false
+ }
+
+ // the local region must have survived the coordinator restart and keep
serving entities
+ awaitAssert(
+ {
+ region ! 2
Review Comment:
nit: this re-sends `2` on every `awaitAssert` retry, so on slow CI several
replies can pile up in `testActor`. The region buffers until the coordinator is
back, so a single `region ! 2` followed by `expectMsg(20.seconds, 2)` should be
enough and is deterministic.
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala:
##########
@@ -966,8 +967,23 @@ abstract class ShardCoordinator(
}: Receive).orElse[Any, Unit](receiveTerminated)
private def terminate(): Unit = {
- if (aliveRegions.exists(_.path.address.hasLocalScope) ||
gracefulShutdownInProgress.exists(
- _.path.address.hasLocalScope)) {
+ val selfIsLeavingCluster = {
+ val selfStatus = cluster.selfMember.status
+ cluster.isTerminated ||
+ CoordinatedShutdown(context.system).shutdownReason().isDefined ||
+ selfStatus == MemberStatus.Leaving ||
+ selfStatus == MemberStatus.Exiting ||
+ selfStatus == MemberStatus.Down
Review Comment:
`Removed`, `PreparingForShutdown` and `ReadyForShutdown` aren't treated as
leaving. During a full cluster shutdown, or after self is `Removed` but before
the Cluster extension is terminated (when the system wasn't stopped via
`CoordinatedShutdown`), a `Terminate` takes the not-leaving path and skips the
drain. The coordinator already tracks full-cluster shutdown in its
`preparingForShutdown` flag, which could be included here. (nit: a `Set` of
leaving statuses would read more easily than the `==` chain.)
##########
cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ClusterShardingCoordinatorLeaseLossSpec.scala:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.net.URLEncoder
+import java.nio.charset.StandardCharsets
+
+import scala.concurrent.duration._
+import scala.util.Success
+
+import org.apache.pekko
+import pekko.cluster.Cluster
+import pekko.cluster.MemberStatus
+import pekko.coordination.lease.TestLease
+import pekko.coordination.lease.TestLeaseExt
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestActors
+import pekko.testkit.TestException
+
+import com.typesafe.config.ConfigFactory
+
+object ClusterShardingCoordinatorLeaseLossSpec {
+ val config = ConfigFactory.parseString("""
+ pekko.loglevel = INFO
+ pekko.actor.provider = cluster
+ pekko.remote.classic.netty.tcp.port = 0
+ pekko.remote.artery.canonical.port = 0
+ pekko.cluster.sharding.coordinator-singleton {
+ use-lease = "test-lease"
+ lease-retry-interval = 200ms
+ }
+ """).withFallback(TestLease.config)
+}
+
+class ClusterShardingCoordinatorLeaseLossSpec
+ extends PekkoSpec(ClusterShardingCoordinatorLeaseLossSpec.config)
+ with ImplicitSender {
+
+ private val cluster = Cluster(system)
+
+ override protected def atStartup(): Unit = {
+ cluster.join(cluster.selfAddress)
+ awaitAssert(cluster.selfMember.status shouldEqual MemberStatus.Up)
+ }
+
+ private val extractEntityId: ShardRegion.ExtractEntityId = {
+ case i: Int => (i.toString, i)
+ }
+
+ private val extractShardId: ShardRegion.ExtractShardId = {
+ case i: Int => (i % 10).toString
+ case _ => "0"
+ }
+
+ "Cluster sharding with a coordinator singleton lease" should {
+
+ "keep the local shard region alive when the coordinator lease is lost" in {
Review Comment:
This covers the lease-loss branch, but nothing checks that the
leaving/shutdown branch still drains the local region before the coordinator
stops, and that branch now has a new gate in front of it. A regression that
made `selfIsLeavingCluster` always `false` would pass here. Could we add a
directional case, e.g. `cluster.leave(selfAddress)` (or start
`CoordinatedShutdown`), wait for the status, then send `Terminate` and assert
the local region gets `GracefulShutdown` before the coordinator stops?
--
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]
