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 a0e44fd96d feat: define custom lease name for sharding and singleton
(#3648)
a0e44fd96d is described below
commit a0e44fd96d3c023c0691d1505cd6898acc30b0c7
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 04:38:43 2026 +0100
feat: define custom lease name for sharding and singleton (#3648)
Motivation:
The lease names used by Cluster Sharding shards and Cluster Singletons are
derived from the
ActorSystem name and the entity type name or singleton actor path. Those
names can become too
long for some lease implementations, such as the Kubernetes lease, and
there was no way to
choose a shorter name.
Modification:
- Add a `leaseName` to `LeaseUsageSettings` (empty means the derived name,
as before), with
`withLeaseName` and the Java `getLeaseName`. The existing two-argument
constructor is kept and
is now public.
- `Shard` uses `<leaseName>-<shardId>` when a custom lease name is defined.
- `ClusterSingletonManager` uses the custom lease name as is when defined.
- The lease name is intentionally not read from config, since every
singleton and entity type
must have a unique lease name; reference.conf documents how to define it
programmatically.
- Document the uniqueness requirement on the `withLeaseSettings` methods of
the classic and
typed sharding and singleton settings.
- Tests in `ShardWithLeaseSpec` and `ClusterSingletonLeaseSpec`.
Result:
Users can define shorter custom lease names for sharding and singletons.
The default lease
names are unchanged.
Tests:
Not run locally - left to CI
References:
Ported from (Apache-2.0 since 2026-10-23):
- https://github.com/akka/akka-core/pull/32122
Notes for reviewers:
- Binary compatibility: only additions (new primary constructor,
`leaseName`, `getLeaseName`,
`withLeaseName`); the existing two-argument constructor is retained. No
MiMa excludes added.
- Wire compatibility: no serialized messages or protobuf changed.
- Config defaults: no new config keys; reference.conf only gains comments.
Default lease names
are unchanged.
- Rolling upgrade: with no custom name, old and new nodes use the same
lease names. If a custom
lease name is introduced, all nodes must use it, so introducing one
during a rolling upgrade
means old and new nodes use different leases; do it with a full cluster
restart.
---
.../sharding/typed/ClusterShardingSettings.scala | 5 ++++
cluster-sharding/src/main/resources/reference.conf | 7 +++++
.../cluster/sharding/ClusterShardingSettings.scala | 5 ++++
.../org/apache/pekko/cluster/sharding/Shard.scala | 12 ++++----
.../cluster/sharding/ShardWithLeaseSpec.scala | 23 ++++++++++++--
cluster-tools/src/main/resources/reference.conf | 6 ++++
.../singleton/ClusterSingletonManager.scala | 16 ++++++----
.../singleton/ClusterSingletonLeaseSpec.scala | 21 +++++++++++++
.../pekko/cluster/typed/ClusterSingleton.scala | 10 +++++++
.../coordination/lease/LeaseUsageSettings.scala | 35 ++++++++++++++++++++--
10 files changed, 125 insertions(+), 15 deletions(-)
diff --git
a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ClusterShardingSettings.scala
b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ClusterShardingSettings.scala
index 2b3933b66f..6e5c554bec 100644
---
a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ClusterShardingSettings.scala
+++
b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/ClusterShardingSettings.scala
@@ -846,6 +846,11 @@ final class ClusterShardingSettings(
def withShardRegionQueryTimeout(duration: java.time.Duration):
ClusterShardingSettings =
copy(shardRegionQueryTimeout = duration.toScala)
+ /**
+ * Note that if you define a custom lease name and have several sharding
entity types each one must have a unique
+ * lease name. If the lease name is undefined it will be derived from
ActorSystem name and shard name,
+ * but that may result in too long lease names.
+ */
def withLeaseSettings(leaseSettings: LeaseUsageSettings) =
copy(leaseSettings = Option(leaseSettings))
/**
diff --git a/cluster-sharding/src/main/resources/reference.conf
b/cluster-sharding/src/main/resources/reference.conf
index 1127d6962c..61bce2fe58 100644
--- a/cluster-sharding/src/main/resources/reference.conf
+++ b/cluster-sharding/src/main/resources/reference.conf
@@ -428,6 +428,13 @@ pekko.cluster.sharding {
# The interval between retries for acquiring the lease
lease-retry-interval = 5s
+ # A custom lease name can be defined with `LeaseUsageSettings.withLeaseName`
in the
+ # leaseSettings of ClusterShardingSettings. The shard identifier is appended
to it.
+ # It is intentionally not defined in config, because if you have several
entity types
+ # each one must have a unique lease name. If undefined, the lease name is
derived from
+ # the ActorSystem name, entity type name and shard identifier, but that may
result in
+ # too long lease names.
+
# Provide a higher level of details in the debug logs, often per routed
message. Be careful about enabling
# in production systems.
verbose-debug-logging = off
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala
index 64b7c0ab6e..2559fc1be1 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala
@@ -974,6 +974,11 @@ final class ClusterShardingSettings(
def withShardRegionQueryTimeout(duration: java.time.Duration):
ClusterShardingSettings =
copy(shardRegionQueryTimeout = duration.toScala)
+ /**
+ * Note that if you define a custom lease name and have several sharding
entity types each one must have a unique
+ * lease name. If the lease name is undefined it will be derived from
ActorSystem name and shard name,
+ * but that may result in too long lease names.
+ */
def withLeaseSettings(leaseSettings: LeaseUsageSettings):
ClusterShardingSettings =
copy(leaseSettings = Some(leaseSettings))
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 57785da53b..560bed9d46 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
@@ -493,11 +493,13 @@ private[pekko] class Shard(
context.system.scheduler.scheduleWithFixedDelay(interval, interval, self,
PassivateIntervalTick)
}
- private val lease = settings.leaseSettings.map(ls =>
- LeaseProvider(context.system).getLease(
- s"${context.system.name}-shard-$typeName-$shardId",
- ls.leaseImplementation,
- Cluster(context.system).selfAddress.hostPort))
+ private val lease = settings.leaseSettings.map { ls =>
+ val leaseName =
+ if (ls.leaseName.isEmpty)
s"${context.system.name}-shard-$typeName-$shardId"
+ else s"${ls.leaseName}-$shardId"
+ LeaseProvider(context.system).getLease(leaseName, ls.leaseImplementation,
+ Cluster(context.system).selfAddress.hostPort)
+ }
private val leaseRetryInterval = settings.leaseSettings match {
case Some(l) => l.leaseRetryInterval
diff --git
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala
index 800d1ffcf9..302dabd54a 100644
---
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala
+++
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ShardWithLeaseSpec.scala
@@ -98,6 +98,20 @@ class ShardWithLeaseSpec extends
PekkoSpec(ShardWithLeaseSpec.config) with WithL
probe.expectMsg("ack hello")
}
+ "use custom lease name with shard id appended" in new Setup {
+ override def customLeaseName: String = "custom-shard-lease"
+ settings.leaseSettings.get.leaseName shouldBe customLeaseName
+ val probe = TestProbe()
+ sharding.tell(EntityEnvelope(1, "hello"), probe.ref)
+ probe.expectNoMessage(shortDuration)
+ leaseFor("1").initialPromise.complete(Success(true))
+ testLeaseExt.getTestLease("custom-shard-lease-1") // lease name with
shard id appended
+ probe.expectMsg("ack hello")
+ intercept[IllegalStateException] {
+ testLeaseExt.getTestLease(s"${system.name}-shard-$typeName-1")
+ }
+ }
+
"retry if lease acquire returns false" in new Setup {
val probe = TestProbe()
val lease =
@@ -153,7 +167,10 @@ class ShardWithLeaseSpec extends
PekkoSpec(ShardWithLeaseSpec.config) with WithL
var typeIdx = 0
trait Setup {
- val settings = ClusterShardingSettings(system).withLeaseSettings(new
LeaseUsageSettings("test-lease", 2.seconds))
+ def customLeaseName: String = ""
+
+ val settings = ClusterShardingSettings(system).withLeaseSettings(
+ new LeaseUsageSettings("test-lease",
2.seconds).withLeaseName(customLeaseName))
// unique type name for each test
val typeName = {
@@ -165,7 +182,9 @@ class ShardWithLeaseSpec extends
PekkoSpec(ShardWithLeaseSpec.config) with WithL
ClusterSharding(system).start(typeName, Props(new EntityActor()),
settings, extractEntityId, extractShardId)
def leaseFor(shardId: ShardId) = awaitAssert {
- val leaseName = s"${system.name}-shard-$typeName-$shardId"
+ val leaseName =
+ if (customLeaseName.isEmpty) s"${system.name}-shard-$typeName-$shardId"
+ else s"$customLeaseName-$shardId"
testLeaseExt.getTestLease(leaseName)
}
}
diff --git a/cluster-tools/src/main/resources/reference.conf
b/cluster-tools/src/main/resources/reference.conf
index cc18387346..e43467bb48 100644
--- a/cluster-tools/src/main/resources/reference.conf
+++ b/cluster-tools/src/main/resources/reference.conf
@@ -191,6 +191,12 @@ pekko.cluster.singleton {
# The interval between retries for acquiring the lease
lease-retry-interval = 5s
+
+ # A custom lease name can be defined with `LeaseUsageSettings.withLeaseName`
in the
+ # leaseSettings of ClusterSingletonManagerSettings. It is intentionally not
defined in
+ # config, because if you have several singletons each one must have a unique
lease name.
+ # If undefined, the lease name is derived from the ActorSystem name and the
singleton actor
+ # path, but that may result in too long lease names.
}
# //#singleton-config
diff --git
a/cluster-tools/src/main/scala/org/apache/pekko/cluster/singleton/ClusterSingletonManager.scala
b/cluster-tools/src/main/scala/org/apache/pekko/cluster/singleton/ClusterSingletonManager.scala
index 6c720e2b11..f7465ac7c1 100644
---
a/cluster-tools/src/main/scala/org/apache/pekko/cluster/singleton/ClusterSingletonManager.scala
+++
b/cluster-tools/src/main/scala/org/apache/pekko/cluster/singleton/ClusterSingletonManager.scala
@@ -153,6 +153,11 @@ final class ClusterSingletonManagerSettings(
def withHandOverRetryInterval(retryInterval: FiniteDuration):
ClusterSingletonManagerSettings =
copy(handOverRetryInterval = retryInterval)
+ /**
+ * Note that if you define a custom lease name and have several singletons
each one must have a unique
+ * lease name. If the lease name is undefined it will be derived from
ActorSystem name and singleton
+ * actor path, but that may result in too long lease names.
+ */
def withLeaseSettings(leaseSettings: LeaseUsageSettings):
ClusterSingletonManagerSettings =
copy(leaseSettings = Some(leaseSettings))
@@ -511,13 +516,14 @@ class ClusterSingletonManager(singletonProps: Props,
terminationMessage: Any, se
role.forall(cluster.selfRoles.contains),
s"This cluster member [${cluster.selfAddress}] doesn't have the role
[$role]")
- private val singletonLeaseName =
s"${context.system.name}-singleton-${self.path}"
-
override val log: MarkerLoggingAdapter = Logging.withMarker(context.system,
this)
- val lease: Option[Lease] = settings.leaseSettings.map(settings =>
- LeaseProvider(context.system)
- .getLease(singletonLeaseName, settings.leaseImplementation,
cluster.selfAddress.hostPort))
+ val lease: Option[Lease] = settings.leaseSettings.map { settings =>
+ val leaseName =
+ if (settings.leaseName.isEmpty)
s"${context.system.name}-singleton-${self.path}"
+ else settings.leaseName
+ LeaseProvider(context.system).getLease(leaseName,
settings.leaseImplementation, cluster.selfAddress.hostPort)
+ }
val leaseRetryInterval: FiniteDuration = settings.leaseSettings match {
case Some(s) => s.leaseRetryInterval
case None => 5.seconds // won't be used
diff --git
a/cluster-tools/src/test/scala/org/apache/pekko/cluster/singleton/ClusterSingletonLeaseSpec.scala
b/cluster-tools/src/test/scala/org/apache/pekko/cluster/singleton/ClusterSingletonLeaseSpec.scala
index c0eafdfa64..aefac3e829 100644
---
a/cluster-tools/src/test/scala/org/apache/pekko/cluster/singleton/ClusterSingletonLeaseSpec.scala
+++
b/cluster-tools/src/test/scala/org/apache/pekko/cluster/singleton/ClusterSingletonLeaseSpec.scala
@@ -196,6 +196,27 @@ class ClusterSingletonLeaseSpec extends
PekkoSpec(ConfigFactory.parseString("""
lifecycleProbe.expectMsg("preStart")
}
+ "use custom lease name" in {
+ val probe = TestProbe()
+ val defaultSettings = nextSettings()
+ val customLeaseName = s"custom-lease-${defaultSettings.singletonName}"
+ val settings =
+
defaultSettings.withLeaseSettings(defaultSettings.leaseSettings.get.withLeaseName(customLeaseName))
+ settings.leaseSettings.get.leaseName shouldBe customLeaseName
+ system.actorOf(
+ ClusterSingletonManager.props(Props(new
ImportantSingleton(probe.ref)), PoisonPill, settings),
+ settings.singletonName)
+ val testLease = awaitAssert {
+ testLeaseExt.getTestLease(customLeaseName)
+ } // allow singleton manager to create the lease
+ probe.expectNoMessage(shortDuration)
+ testLease.initialPromise.complete(Success(true))
+ probe.expectMsg("preStart")
+ intercept[IllegalStateException] {
+ testLeaseExt.getTestLease(leaseNameFor(settings))
+ }
+ }
+
"release lease when leaving oldest" in {
val singletonProbe = TestProbe()
val settings = nextSettings()
diff --git
a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala
b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala
index 1a3b122a6c..d1858b0e1b 100644
---
a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala
+++
b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala
@@ -84,6 +84,11 @@ final class ClusterSingletonSettings(
def withBufferSize(bufferSize: Int): ClusterSingletonSettings =
copy(bufferSize = bufferSize)
+ /**
+ * Note that if you define a custom lease name and have several singletons
each one must have a unique
+ * lease name. If the lease name is undefined it will be derived from
ActorSystem name and singleton
+ * actor path, but that may result in too long lease names.
+ */
def withLeaseSettings(leaseSettings: LeaseUsageSettings) =
copy(leaseSettings = Option(leaseSettings))
private def copy(
@@ -312,6 +317,11 @@ final class ClusterSingletonManagerSettings(
def withHandOverRetryInterval(retryInterval: java.time.Duration):
ClusterSingletonManagerSettings =
withHandOverRetryInterval(retryInterval.toScala)
+ /**
+ * Note that if you define a custom lease name and have several singletons
each one must have a unique
+ * lease name. If the lease name is undefined it will be derived from
ActorSystem name and singleton
+ * actor path, but that may result in too long lease names.
+ */
def withLeaseSettings(leaseSettings: LeaseUsageSettings) =
copy(leaseSettings = Option(leaseSettings))
private def copy(
diff --git
a/coordination/src/main/scala/org/apache/pekko/coordination/lease/LeaseUsageSettings.scala
b/coordination/src/main/scala/org/apache/pekko/coordination/lease/LeaseUsageSettings.scala
index 8fbae9a8eb..ef7bb130d2 100644
---
a/coordination/src/main/scala/org/apache/pekko/coordination/lease/LeaseUsageSettings.scala
+++
b/coordination/src/main/scala/org/apache/pekko/coordination/lease/LeaseUsageSettings.scala
@@ -19,9 +19,38 @@ import scala.jdk.DurationConverters._
import org.apache.pekko
import pekko.util.PrettyDuration._
-final class LeaseUsageSettings private[pekko] (val leaseImplementation: String,
- val leaseRetryInterval: FiniteDuration) {
+/**
+ * Note that if you define a custom lease name and have several Cluster
Singletons or Cluster Sharding
+ * entity types each one must have a unique lease name. If the lease name is
undefined it will be derived
+ * from ActorSystem name and other component names, but that may result in too
long lease names.
+ */
+final class LeaseUsageSettings private[pekko] (
+ val leaseImplementation: String,
+ val leaseRetryInterval: FiniteDuration,
+ val leaseName: String) {
+
+ def this(leaseImplementation: String, leaseRetryInterval: FiniteDuration) =
+ this(leaseImplementation, leaseRetryInterval, leaseName = "")
+
def getLeaseRetryInterval(): java.time.Duration = leaseRetryInterval.toJava
- override def toString = s"LeaseUsageSettings($leaseImplementation,
${leaseRetryInterval.pretty})"
+ /**
+ * Java API: The custom lease name, empty if the lease name is derived from
the
+ * ActorSystem name and other component names.
+ */
+ def getLeaseName(): String = leaseName
+
+ /**
+ * Define a custom lease name. Note that if you have several Cluster
Singletons or Cluster Sharding
+ * entity types each one must have a unique lease name. For Cluster Sharding
the shard identifier
+ * is appended to the custom lease name, so that each shard has its own
lease.
+ * If the lease name is undefined (empty) it will be derived from
ActorSystem name and other
+ * component names, but that may result in too long lease names.
+ */
+ def withLeaseName(name: String): LeaseUsageSettings =
+ new LeaseUsageSettings(leaseImplementation, leaseRetryInterval, name)
+
+ override def toString =
+ if (leaseName.isEmpty) s"LeaseUsageSettings($leaseImplementation,
${leaseRetryInterval.pretty})"
+ else s"LeaseUsageSettings($leaseImplementation,
${leaseRetryInterval.pretty}, $leaseName)"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
