This is an automated email from the ASF dual-hosted git repository.
pjfanning 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 17fb00e51e consistent-hashing shard allocation strategy (#3270)
17fb00e51e is described below
commit 17fb00e51e847997a31cf6f130545ba3893285b8
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Jul 6 18:42:18 2026 +0100
consistent-hashing shard allocation strategy (#3270)
* Apply remaining changes
* manual changes
* more
* more
* more
* compile issues
* build issues
* unused imports
* docs: add Colocate Shards and EventsBySliceFirehoseQuery documentation
(#57)
Motivation:
The consistent-hashing shard allocation and EventsBySliceFirehoseQuery code
were ported in this PR, but the corresponding user-facing documentation from
the upstream commits was not yet included.
Modification:
- Add the "Colocate Shards" section to typed/cluster-sharding.md (from akka
522f566128), wired to the existing
ConsistentHashingShardAllocationCompileOnly
examples.
- Add the EventsBySliceFirehoseQuery scalability paragraph to
persistence-query.md and a `See also [[EventsBySliceFirehoseQuery]]`
cross-reference to both EventsBySliceQuery DSL traits (from akka
362ab9301e).
Result:
The Cluster Sharding and Persistence Query docs describe the new
consistent-hashing colocation and firehose query features.
Tests: Not run - docs only (sbt "docs/paradox" builds successfully)
References:
Ports docs from akka/akka-core@522f566128 (akka#31967) and
akka/akka-core@362ab9301e (akka#31957), which are now Apache licensed.
Completes documentation for apache/pekko#3270.
* fix: update ConsistentHashingShardAllocationStrategySpec for pekko
protocol hash (#58)
* fix: update ConsistentHashingShardAllocationStrategySpec for pekko
protocol
Motivation:
ConsistentHashingShardAllocationStrategySpec was ported from Akka 2.8.3
with test expectations based on the akka:// protocol. Since Pekko uses
pekko:// as the protocol name, Address.toString produces different
strings, causing ConsistentHash to compute different hash values and
map shards to different nodes.
Modification:
Updated all test expectations to match the consistent hash results
computed with pekko:// addresses. Key changes:
- 3-node: shard "2" maps to regionA (was regionB), shard "3" maps to
regionA (was regionC)
- 4-node: shard "0" maps to regionD (was regionC), shard "2" to regionA
(was regionD)
- 2-node: shard "2" maps to regionA (was regionB), shard "3" to regionA
(was regionB)
- Rebalance assertions updated to reflect the new 4-node shard distribution
Result:
All 10 ConsistentHashingShardAllocationStrategySpec tests pass.
No changes to production code.
Tests:
sbt "cluster-sharding / Test / testOnly
org.apache.pekko.cluster.sharding.ConsistentHashingShardAllocationStrategySpec"
Tests: succeeded 10, failed 0
References:
Fixes apache/pekko#3270
* fix: update stale comment in ConsistentHashingShardAllocationStrategySpec
In Pekko's consistent hash, shards 2, 3, 10 and 14 are all already at
the right place (not just 10 and 14 as in Akka).
* scalafmt
* Update consistent-hashing-allocation.excludes
* Update ConsistentHashingShardAllocationCompileOnlyTest.java
---------
Co-authored-by: copilot-swe-agent[bot]
<[email protected]>
Co-authored-by: He-Pin(kerr) <[email protected]>
---
...stentHashingShardAllocationCompileOnlyTest.java | 141 +++++++++++
...tentHashingShardAllocationCompileOnlySpec.scala | 94 ++++++++
.../consistent-hashing-allocation.excludes | 22 ++
.../ConsistentHashingShardAllocationStrategy.scala | 163 +++++++++++++
.../pekko/cluster/sharding/ShardCoordinator.scala | 5 +-
.../AbstractLeastShardAllocationStrategy.scala | 87 ++-----
...egy.scala => ClusterShardAllocationMixin.scala} | 71 ++----
.../internal/LeastShardAllocationStrategy.scala | 4 +-
...sistentHashingShardAllocationStrategySpec.scala | 257 +++++++++++++++++++++
...eprecatedLeastShardAllocationStrategySpec.scala | 24 +-
.../LeastShardAllocationStrategySpec.scala | 23 +-
docs/src/main/paradox/persistence-query.md | 5 +
.../typed/cluster-sharded-daemon-process.md | 38 ++-
docs/src/main/paradox/typed/cluster-sharding.md | 38 +++
14 files changed, 813 insertions(+), 159 deletions(-)
diff --git
a/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlyTest.java
b/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlyTest.java
new file mode 100644
index 0000000000..3a002220af
--- /dev/null
+++
b/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlyTest.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package jdocs.org.apache.pekko.cluster.sharding.typed;
+
+import org.apache.pekko.actor.typed.ActorSystem;
+import org.apache.pekko.actor.typed.Behavior;
+import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
+import org.apache.pekko.actor.typed.javadsl.ActorContext;
+import org.apache.pekko.actor.typed.javadsl.Behaviors;
+import org.apache.pekko.actor.typed.javadsl.Receive;
+import
org.apache.pekko.cluster.sharding.ConsistentHashingShardAllocationStrategy;
+import org.apache.pekko.cluster.sharding.typed.ShardingEnvelope;
+import org.apache.pekko.cluster.sharding.typed.ShardingMessageExtractor;
+import org.apache.pekko.cluster.sharding.typed.javadsl.ClusterSharding;
+import org.apache.pekko.cluster.sharding.typed.javadsl.Entity;
+import org.apache.pekko.cluster.sharding.typed.javadsl.EntityTypeKey;
+
+public class ConsistentHashingShardAllocationCompileOnlyTest {
+
+ // #building
+ public static class Building extends AbstractBehavior<Building.Command> {
+
+ static int NUMBER_OF_SHARDS = 100;
+
+ static final class MessageExtractor
+ extends ShardingMessageExtractor<ShardingEnvelope<Building.Command>,
Building.Command> {
+ @Override
+ public String entityId(ShardingEnvelope<Command> envelope) {
+ return envelope.entityId();
+ }
+
+ @Override
+ public String shardId(String entityId) {
+ return String.valueOf(Math.abs(entityId.hashCode() %
NUMBER_OF_SHARDS));
+ }
+
+ @Override
+ public Command unwrapMessage(ShardingEnvelope<Command> envelope) {
+ return envelope.message();
+ }
+ }
+
+ static final EntityTypeKey<Building.Command> typeKey =
+ EntityTypeKey.create(Building.Command.class, "Building");
+
+ public interface Command {}
+
+ public static Behavior<Building.Command> create(String entityId) {
+ return Behaviors.setup(context -> new Building(context, entityId));
+ }
+
+ private Building(ActorContext<Building.Command> context, String entityId) {
+ super(context);
+ }
+
+ @Override
+ public Receive<Building.Command> createReceive() {
+ return newReceiveBuilder().build();
+ }
+ }
+
+ // #building
+
+ // #device
+ public static class Device extends AbstractBehavior<Device.Command> {
+
+ static final class MessageExtractor
+ extends ShardingMessageExtractor<ShardingEnvelope<Device.Command>,
Device.Command> {
+ @Override
+ public String entityId(ShardingEnvelope<Command> envelope) {
+ return envelope.entityId();
+ }
+
+ @Override
+ public String shardId(String entityId) {
+ // Use same shardId as the Building to colocate Building and Device
+ // we have the buildingId as prefix in the entityId
+ String buildingId = entityId.split(":")[0];
+ return String.valueOf(Math.abs(buildingId.hashCode() %
Building.NUMBER_OF_SHARDS));
+ }
+
+ @Override
+ public Command unwrapMessage(ShardingEnvelope<Command> envelope) {
+ return envelope.message();
+ }
+ }
+
+ static final EntityTypeKey<Device.Command> typeKey =
+ EntityTypeKey.create(Device.Command.class, "Device");
+
+ public interface Command {}
+
+ public static Behavior<Device.Command> create(String entityId) {
+ return Behaviors.setup(context -> new Device(context, entityId));
+ }
+
+ private Device(ActorContext<Device.Command> context, String entityId) {
+ super(context);
+ }
+
+ @Override
+ public Receive<Device.Command> createReceive() {
+ return newReceiveBuilder().build();
+ }
+ }
+
+ // #device
+
+ void example() {
+ ActorSystem<?> system = null;
+
+ // #init
+ int rebalanceLimit = 10;
+
+ ClusterSharding.get(system)
+ .init(
+ Entity.of(Building.typeKey, ctx ->
Building.create(ctx.getEntityId()))
+ .withMessageExtractor(new Building.MessageExtractor())
+ .withAllocationStrategy(
+ new
ConsistentHashingShardAllocationStrategy(rebalanceLimit)));
+
+ ClusterSharding.get(system)
+ .init(
+ Entity.of(Device.typeKey, ctx -> Device.create(ctx.getEntityId()))
+ .withMessageExtractor(new Device.MessageExtractor())
+ .withAllocationStrategy(
+ new
ConsistentHashingShardAllocationStrategy(rebalanceLimit)));
+ // #init
+ }
+}
diff --git
a/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlySpec.scala
b/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlySpec.scala
new file mode 100644
index 0000000000..ea92ce517b
--- /dev/null
+++
b/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlySpec.scala
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package docs.org.apache.pekko.cluster.sharding.typed
+
+import org.apache.pekko
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.cluster.sharding.ConsistentHashingShardAllocationStrategy
+import pekko.cluster.sharding.typed.ShardingEnvelope
+import pekko.cluster.sharding.typed.ShardingMessageExtractor
+import pekko.cluster.sharding.typed.scaladsl.ClusterSharding
+import pekko.cluster.sharding.typed.scaladsl.Entity
+import pekko.cluster.sharding.typed.scaladsl.EntityTypeKey
+
+class ConsistentHashingShardAllocationCompileOnlySpec {
+
+ // #building
+ object Building {
+ val TypeKey = EntityTypeKey[Command]("Building")
+
+ val NumberOfShards = 100
+
+ final class MessageExtractor extends
ShardingMessageExtractor[ShardingEnvelope[Command], Command] {
+
+ override def entityId(envelope: ShardingEnvelope[Command]): String =
+ envelope.entityId
+
+ override def shardId(entityId: String): String =
+ math.abs(entityId.hashCode % NumberOfShards).toString
+
+ override def unwrapMessage(envelope: ShardingEnvelope[Command]): Command
=
+ envelope.message
+ }
+
+ sealed trait Command
+
+ def apply(entityId: String): Behavior[Command] = ???
+ }
+
+ // #building
+
+ // #device
+ object Device {
+ val TypeKey = EntityTypeKey[Command]("Device")
+
+ final class MessageExtractor extends
ShardingMessageExtractor[ShardingEnvelope[Command], Command] {
+
+ override def entityId(envelope: ShardingEnvelope[Command]): String =
+ envelope.entityId
+
+ override def shardId(entityId: String): String = {
+ // Use same shardId as the Building to colocate Building and Device
+ // we have the buildingId as prefix in the entityId
+ val buildingId = entityId.split(':').head
+ math.abs(buildingId.hashCode % Building.NumberOfShards).toString
+ }
+
+ override def unwrapMessage(envelope: ShardingEnvelope[Command]): Command
=
+ envelope.message
+ }
+
+ sealed trait Command
+
+ def apply(entityId: String): Behavior[Command] = ???
+ }
+
+ // #device
+
+ val system: ActorSystem[?] = ???
+
+ // #init
+ ClusterSharding(system).init(
+ Entity(Building.TypeKey)(createBehavior = entityContext =>
Building(entityContext.entityId))
+ .withMessageExtractor(new Building.MessageExtractor)
+ .withAllocationStrategy(new
ConsistentHashingShardAllocationStrategy(rebalanceLimit = 10)))
+
+ ClusterSharding(system).init(
+ Entity(Device.TypeKey)(createBehavior = entityContext =>
Device(entityContext.entityId))
+ .withMessageExtractor(new Device.MessageExtractor)
+ .withAllocationStrategy(new
ConsistentHashingShardAllocationStrategy(rebalanceLimit = 10)))
+ // #init
+
+}
diff --git
a/cluster-sharding/src/main/mima-filters/2.0.x.backwards.excludes/consistent-hashing-allocation.excludes
b/cluster-sharding/src/main/mima-filters/2.0.x.backwards.excludes/consistent-hashing-allocation.excludes
new file mode 100644
index 0000000000..f5326a3be4
--- /dev/null
+++
b/cluster-sharding/src/main/mima-filters/2.0.x.backwards.excludes/consistent-hashing-allocation.excludes
@@ -0,0 +1,22 @@
+# 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.
+
+# Internal classes affected by refactor related to Colocating shards with
consistent hashing
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy$")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy$RegionEntry")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy$RegionEntry$")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy$ShardSuitabilityOrdering$")
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ConsistentHashingShardAllocationStrategy.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ConsistentHashingShardAllocationStrategy.scala
new file mode 100644
index 0000000000..0778c519db
--- /dev/null
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ConsistentHashingShardAllocationStrategy.scala
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding
+
+import scala.annotation.nowarn
+import scala.collection.immutable
+import scala.concurrent.Future
+
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.actor.ActorSystem
+import pekko.actor.Address
+import pekko.annotation.DoNotInherit
+import pekko.cluster.Cluster
+import pekko.cluster.ClusterEvent.CurrentClusterState
+import pekko.cluster.Member
+import
pekko.cluster.sharding.ShardCoordinator.ActorSystemDependentAllocationStrategy
+import pekko.cluster.sharding.ShardRegion.ShardId
+import pekko.cluster.sharding.internal.ClusterShardAllocationMixin
+import
pekko.cluster.sharding.internal.ClusterShardAllocationMixin.ShardSuitabilityOrdering
+import pekko.event.Logging
+import pekko.event.LoggingAdapter
+import pekko.routing.ConsistentHash
+
+object ConsistentHashingShardAllocationStrategy {
+ private val emptyRebalanceResult = Future.successful(Set.empty[ShardId])
+}
+
+/**
+ * [[pekko.cluster.sharding.ShardCoordinator.ShardAllocationStrategy]] that is
using consistent
+ * hashing. This can be useful when shards with the same shard id for
different entity types
+ * should be best effort colocated to the same nodes.
+ *
+ * When adding or removing nodes it will rebalance according to the new
consistent hashing,
+ * but that means that only a few shards will be rebalanced and others remain
on the same
+ * location.
+ *
+ * A good explanation of Consistent Hashing:
+ * https://tom-e-white.com/2007/11/consistent-hashing.html
+ *
+ * Create a new instance of this for each entity types, i.e. a
`ConsistentHashingShardAllocationStrategy`
+ * instance must not be shared between different entity types.
+ *
+ * Not intended for public inheritance/implementation
+ */
+@DoNotInherit
+class ConsistentHashingShardAllocationStrategy(rebalanceLimit: Int)
+ extends ActorSystemDependentAllocationStrategy
+ with ClusterShardAllocationMixin {
+ import ConsistentHashingShardAllocationStrategy.emptyRebalanceResult
+
+ private var cluster: Cluster = _
+ private var _log: LoggingAdapter = _
+
+ private val virtualNodesFactor = 10
+ private var hashedByNodes: Vector[Address] = Vector.empty
+ private var consistentHashing: ConsistentHash[Address] = ConsistentHash(Nil,
virtualNodesFactor)
+
+ override def start(system: ActorSystem): Unit = {
+ cluster = Cluster(system)
+ _log = Logging(system, classOf[ConsistentHashingShardAllocationStrategy])
+ }
+
+ // can be overridden for tests without real Cluster, i.e. without
`start(system)`
+ protected def log: LoggingAdapter =
+ _log
+
+ override protected def clusterState: CurrentClusterState = cluster.state
+ override protected def selfMember: Member = cluster.selfMember
+
+ @nowarn("msg=never used")
+ override def allocateShard(
+ requester: ActorRef,
+ shardId: ShardId,
+ currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardId]]):
Future[ActorRef] = {
+ val nodes = nodesForRegions(currentShardAllocations)
+ updateHashing(nodes)
+ val node = consistentHashing.nodeFor(shardId)
+ currentShardAllocations.keysIterator.find(region => nodeForRegion(region)
== node) match {
+ case Some(region) => Future.successful(region)
+ case None =>
+ throw new IllegalStateException(s"currentShardAllocations should
include region for node [$node]")
+ }
+ }
+
+ override def rebalance(
+ currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardId]],
+ rebalanceInProgress: Set[ShardId]): Future[Set[ShardId]] = {
+
+ val sortedRegionEntries =
regionEntriesFor(currentShardAllocations).toVector.sorted(ShardSuitabilityOrdering)
+ if (!isAGoodTimeToRebalance(sortedRegionEntries)) {
+ emptyRebalanceResult
+ } else {
+
+ val nodes = nodesForRegions(currentShardAllocations)
+ updateHashing(nodes)
+
+ val regionByNode = currentShardAllocations.keysIterator.map(region =>
nodeForRegion(region) -> region).toMap
+
+ var result = Set.empty[String]
+
+ def lessThanLimit: Boolean =
+ rebalanceLimit <= 0 || result.size < rebalanceLimit
+
+ currentShardAllocations
+ // deterministic order, at least easier to test
+ .toVector.sortBy { case (region, _) => nodeForRegion(region)
}(Address.addressOrdering).foreach {
+ case (currentRegion, shardIds) =>
+ shardIds.foreach { shardId =>
+ if (lessThanLimit && !rebalanceInProgress.contains(shardId)) {
+ val node = consistentHashing.nodeFor(shardId)
+ regionByNode.get(node) match {
+ case Some(region) =>
+ if (region != currentRegion) {
+ log.debug(
+ "Rebalance needed for shard [{}], from [{}] to [{}]",
+ shardId,
+ nodeForRegion(currentRegion),
+ node)
+ result += shardId
+ }
+ case None =>
+ throw new IllegalStateException(s"currentShardAllocations
should include region for node [$node]")
+ }
+ }
+ }
+ }
+
+ Future.successful(result)
+ }
+ }
+
+ private def nodesForRegions(
+ currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardId]]):
Vector[Address] = {
+ currentShardAllocations.keysIterator.map(nodeForRegion).toVector
+ }
+
+ private def nodeForRegion(region: ActorRef): Address =
+ if (region.path.address.hasLocalScope) selfMember.address
+ else region.path.address
+
+ private def updateHashing(nodes: Vector[Address]): Unit = {
+ val sortedNodes = nodes.sorted
+ if (sortedNodes != hashedByNodes) {
+ if (log.isDebugEnabled)
+ log.debug("Update consistent hashing nodes [{}]",
sortedNodes.mkString(", "))
+ hashedByNodes = sortedNodes
+ consistentHashing = ConsistentHash(sortedNodes, virtualNodesFactor)
+ }
+ }
+
+}
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
index b353bd7ede..1d9baa46dc 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
@@ -39,7 +39,8 @@ import pekko.cluster.sharding.internal.{
RememberEntitiesProvider
}
import pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy
-import
pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy.RegionEntry
+import pekko.cluster.sharding.internal.ClusterShardAllocationMixin.RegionEntry
+import
pekko.cluster.sharding.internal.ClusterShardAllocationMixin.ShardSuitabilityOrdering
import
pekko.cluster.sharding.internal.EventSourcedRememberEntitiesCoordinatorStore.MigrationMarker
import pekko.event.{ BusLogging, Logging }
import pekko.pattern.{ pipe, AskTimeoutException }
@@ -296,8 +297,6 @@ object ShardCoordinator {
extends AbstractLeastShardAllocationStrategy
with Serializable {
- import AbstractLeastShardAllocationStrategy.ShardSuitabilityOrdering
-
override def rebalance(
currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardId]],
rebalanceInProgress: Set[ShardId]): Future[Set[ShardId]] = {
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/AbstractLeastShardAllocationStrategy.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/AbstractLeastShardAllocationStrategy.scala
index 433bbbd49a..1826e83f5b 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/AbstractLeastShardAllocationStrategy.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/AbstractLeastShardAllocationStrategy.scala
@@ -13,8 +13,6 @@
package org.apache.pekko.cluster.sharding.internal
-import java.lang.{ Boolean => JBoolean, Integer => JInteger }
-
import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration.DurationInt
@@ -22,57 +20,35 @@ import scala.concurrent.duration.DurationInt
import org.apache.pekko
import pekko.actor.ActorRef
import pekko.actor.ActorSystem
-import pekko.actor.Address
import pekko.annotation.InternalApi
import pekko.cluster.Cluster
import pekko.cluster.ClusterEvent.CurrentClusterState
import pekko.cluster.Member
-import pekko.cluster.MemberStatus
import
pekko.cluster.sharding.ShardCoordinator.ActorSystemDependentAllocationStrategy
import pekko.cluster.sharding.ShardRegion.ShardId
+import pekko.cluster.sharding.internal.ClusterShardAllocationMixin.RegionEntry
+import
pekko.cluster.sharding.internal.ClusterShardAllocationMixin.ShardSuitabilityOrdering
import pekko.pattern.after
/**
- * Common logic for the least shard allocation strategy implementations
- *
* INTERNAL API
*/
@InternalApi
private[pekko] object AbstractLeastShardAllocationStrategy {
- import MemberStatus._
- private val JoiningCluster: Set[MemberStatus] = Set(Joining, WeaklyUp)
- private val LeavingClusterStatuses: Set[MemberStatus] = Set(Leaving,
Exiting, Down)
-
- type AllocationMap = Map[ActorRef, immutable.IndexedSeq[ShardId]]
-
- final case class RegionEntry(region: ActorRef, member: Member, shardIds:
immutable.IndexedSeq[ShardId])
-
- implicit object ShardSuitabilityOrdering extends Ordering[RegionEntry] {
- override def compare(x: RegionEntry, y: RegionEntry): Int = {
- val RegionEntry(_, memberX, allocatedShardsX) = x
- val RegionEntry(_, memberY, allocatedShardsY) = y
- if (memberX.status != memberY.status) {
- // prefer allocating to nodes that are not on their way out of the
cluster
- val xIsLeaving = LeavingClusterStatuses(memberX.status)
- val yIsLeaving = LeavingClusterStatuses(memberY.status)
- JBoolean.compare(xIsLeaving, yIsLeaving)
- } else if (memberX.appVersion != memberY.appVersion) {
- // prefer nodes with the highest rolling update app version
- memberY.appVersion.compareTo(memberX.appVersion)
- } else {
- // prefer the node with the least allocated shards
- JInteger.compare(allocatedShardsX.size, allocatedShardsY.size)
- }
- }
- }
+ // kept for binary compatibility - delegate to ClusterShardAllocationMixin
+ type AllocationMap = ClusterShardAllocationMixin.AllocationMap
+ val ShardSuitabilityOrdering: Ordering[RegionEntry] =
ClusterShardAllocationMixin.ShardSuitabilityOrdering
+ type RegionEntry = ClusterShardAllocationMixin.RegionEntry
+ val RegionEntry: ClusterShardAllocationMixin.RegionEntry.type =
ClusterShardAllocationMixin.RegionEntry
}
/**
- * INTERNAL API
+ * INTERNAL API: Common logic for the least shard allocation strategy
implementations
*/
@InternalApi
-private[pekko] abstract class AbstractLeastShardAllocationStrategy extends
ActorSystemDependentAllocationStrategy {
- import AbstractLeastShardAllocationStrategy._
+private[pekko] abstract class AbstractLeastShardAllocationStrategy
+ extends ActorSystemDependentAllocationStrategy
+ with ClusterShardAllocationMixin {
@volatile private var system: ActorSystem = _
@volatile private var cluster: Cluster = _
@@ -83,8 +59,8 @@ private[pekko] abstract class
AbstractLeastShardAllocationStrategy extends Actor
}
// protected for testability
- protected def clusterState: CurrentClusterState = cluster.state
- protected def selfMember: Member = cluster.selfMember
+ override protected def clusterState: CurrentClusterState = cluster.state
+ override protected def selfMember: Member = cluster.selfMember
override def allocateShard(
requester: ActorRef,
@@ -101,47 +77,10 @@ private[pekko] abstract class
AbstractLeastShardAllocationStrategy extends Actor
}
}
- final protected def isAGoodTimeToRebalance(regionEntries:
Iterable[RegionEntry]): Boolean = {
- // Avoid rebalance when rolling update is in progress
- // (This will ignore versions on members with no shard regions, because of
sharding role or not yet completed joining)
- regionEntries.headOption match {
- case None => false // empty list of regions, probably not a
good time to rebalance...
- case Some(firstRegion) =>
- def allNodesSameVersion = {
- regionEntries.forall(_.member.appVersion ==
firstRegion.member.appVersion)
- }
- // Rebalance requires ack from regions and proxies - no need to
rebalance if it cannot be completed
- // FIXME #29589, we currently only look at same dc but proxies in
other dcs may delay complete as well right now
- def neededMembersReachable =
- !clusterState.members.exists(m => m.dataCenter ==
selfMember.dataCenter && clusterState.unreachable(m))
- // No members in same dc joining, we want that to complete before
rebalance, such nodes should reach Up soon
- def membersInProgressOfJoining =
- clusterState.members.exists(m => m.dataCenter ==
selfMember.dataCenter && JoiningCluster(m.status))
-
- allNodesSameVersion && neededMembersReachable &&
!membersInProgressOfJoining
- }
- }
-
final protected def mostSuitableRegion(
regionEntries: Iterable[RegionEntry]): (ActorRef,
immutable.IndexedSeq[ShardId]) = {
val mostSuitableEntry = regionEntries.min(ShardSuitabilityOrdering)
mostSuitableEntry.region -> mostSuitableEntry.shardIds
}
- final protected def regionEntriesFor(currentShardAllocations:
AllocationMap): Iterable[RegionEntry] = {
- val addressToMember: Map[Address, Member] =
clusterState.members.iterator.map(m => m.address -> m).toMap
- currentShardAllocations.flatMap {
- case (region, shardIds) =>
- val regionAddress = {
- if (region.path.address.hasLocalScope) selfMember.address
- else region.path.address
- }
-
- val memberForRegion = addressToMember.get(regionAddress)
- // if the member is unknown (very unlikely but not impossible) because
of view not updated yet
- // that node is ignored for this invocation
- memberForRegion.map(member => RegionEntry(region, member, shardIds))
- }
- }
-
}
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/AbstractLeastShardAllocationStrategy.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/ClusterShardAllocationMixin.scala
similarity index 63%
copy from
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/AbstractLeastShardAllocationStrategy.scala
copy to
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/ClusterShardAllocationMixin.scala
index 433bbbd49a..ec23b34ce9 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/AbstractLeastShardAllocationStrategy.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/ClusterShardAllocationMixin.scala
@@ -8,43 +8,39 @@
*/
/*
- * Copyright (C) 2020-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2020-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.cluster.sharding.internal
-import java.lang.{ Boolean => JBoolean, Integer => JInteger }
+import java.lang.{ Boolean => JBoolean }
+import java.lang.{ Integer => JInteger }
import scala.collection.immutable
-import scala.concurrent.Future
-import scala.concurrent.duration.DurationInt
import org.apache.pekko
import pekko.actor.ActorRef
-import pekko.actor.ActorSystem
import pekko.actor.Address
import pekko.annotation.InternalApi
-import pekko.cluster.Cluster
import pekko.cluster.ClusterEvent.CurrentClusterState
import pekko.cluster.Member
import pekko.cluster.MemberStatus
-import
pekko.cluster.sharding.ShardCoordinator.ActorSystemDependentAllocationStrategy
import pekko.cluster.sharding.ShardRegion.ShardId
-import pekko.pattern.after
+import
pekko.cluster.sharding.internal.ClusterShardAllocationMixin.JoiningCluster
+import pekko.cluster.sharding.internal.ClusterShardAllocationMixin.RegionEntry
/**
- * Common logic for the least shard allocation strategy implementations
- *
* INTERNAL API
*/
-@InternalApi
-private[pekko] object AbstractLeastShardAllocationStrategy {
- import MemberStatus._
- private val JoiningCluster: Set[MemberStatus] = Set(Joining, WeaklyUp)
- private val LeavingClusterStatuses: Set[MemberStatus] = Set(Leaving,
Exiting, Down)
+@InternalApi private[pekko] object ClusterShardAllocationMixin {
type AllocationMap = Map[ActorRef, immutable.IndexedSeq[ShardId]]
+ import MemberStatus._
+
+ val JoiningCluster: Set[MemberStatus] = Set(Joining, WeaklyUp)
+ val LeavingClusterStatuses: Set[MemberStatus] = Set(Leaving, Exiting, Down)
+
final case class RegionEntry(region: ActorRef, member: Member, shardIds:
immutable.IndexedSeq[ShardId])
implicit object ShardSuitabilityOrdering extends Ordering[RegionEntry] {
@@ -65,41 +61,18 @@ private[pekko] object AbstractLeastShardAllocationStrategy {
}
}
}
+
}
/**
* INTERNAL API
*/
-@InternalApi
-private[pekko] abstract class AbstractLeastShardAllocationStrategy extends
ActorSystemDependentAllocationStrategy {
- import AbstractLeastShardAllocationStrategy._
+@InternalApi private[pekko] trait ClusterShardAllocationMixin {
+ import ClusterShardAllocationMixin.AllocationMap
- @volatile private var system: ActorSystem = _
- @volatile private var cluster: Cluster = _
-
- override def start(system: ActorSystem): Unit = {
- this.system = system
- cluster = Cluster(system)
- }
+ protected def clusterState: CurrentClusterState
- // protected for testability
- protected def clusterState: CurrentClusterState = cluster.state
- protected def selfMember: Member = cluster.selfMember
-
- override def allocateShard(
- requester: ActorRef,
- shardId: ShardId,
- currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardId]]):
Future[ActorRef] = {
- val regionEntries = regionEntriesFor(currentShardAllocations)
- if (regionEntries.isEmpty) {
- // very unlikely to ever happen but possible because of cluster state
view not yet updated when collecting
- // region entries, view should be updated after a very short time
- after(50.millis)(allocateShard(requester, shardId,
currentShardAllocations))(system)
- } else {
- val (region, _) = mostSuitableRegion(regionEntries)
- Future.successful(region)
- }
- }
+ protected def selfMember: Member
final protected def isAGoodTimeToRebalance(regionEntries:
Iterable[RegionEntry]): Boolean = {
// Avoid rebalance when rolling update is in progress
@@ -107,13 +80,14 @@ private[pekko] abstract class
AbstractLeastShardAllocationStrategy extends Actor
regionEntries.headOption match {
case None => false // empty list of regions, probably not a
good time to rebalance...
case Some(firstRegion) =>
- def allNodesSameVersion = {
+ def allNodesSameVersion =
regionEntries.forall(_.member.appVersion ==
firstRegion.member.appVersion)
- }
+
// Rebalance requires ack from regions and proxies - no need to
rebalance if it cannot be completed
// FIXME #29589, we currently only look at same dc but proxies in
other dcs may delay complete as well right now
def neededMembersReachable =
!clusterState.members.exists(m => m.dataCenter ==
selfMember.dataCenter && clusterState.unreachable(m))
+
// No members in same dc joining, we want that to complete before
rebalance, such nodes should reach Up soon
def membersInProgressOfJoining =
clusterState.members.exists(m => m.dataCenter ==
selfMember.dataCenter && JoiningCluster(m.status))
@@ -122,12 +96,6 @@ private[pekko] abstract class
AbstractLeastShardAllocationStrategy extends Actor
}
}
- final protected def mostSuitableRegion(
- regionEntries: Iterable[RegionEntry]): (ActorRef,
immutable.IndexedSeq[ShardId]) = {
- val mostSuitableEntry = regionEntries.min(ShardSuitabilityOrdering)
- mostSuitableEntry.region -> mostSuitableEntry.shardIds
- }
-
final protected def regionEntriesFor(currentShardAllocations:
AllocationMap): Iterable[RegionEntry] = {
val addressToMember: Map[Address, Member] =
clusterState.members.iterator.map(m => m.address -> m).toMap
currentShardAllocations.flatMap {
@@ -143,5 +111,4 @@ private[pekko] abstract class
AbstractLeastShardAllocationStrategy extends Actor
memberForRegion.map(member => RegionEntry(region, member, shardIds))
}
}
-
}
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala
index 86be1f4481..61d7ba5dd0 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala
@@ -20,8 +20,8 @@ import org.apache.pekko
import pekko.actor.ActorRef
import pekko.annotation.InternalApi
import pekko.cluster.sharding.ShardRegion.ShardId
-import
pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy.RegionEntry
-import
pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy.ShardSuitabilityOrdering
+import pekko.cluster.sharding.internal.ClusterShardAllocationMixin.RegionEntry
+import
pekko.cluster.sharding.internal.ClusterShardAllocationMixin.ShardSuitabilityOrdering
/**
* INTERNAL API
diff --git
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ConsistentHashingShardAllocationStrategySpec.scala
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ConsistentHashingShardAllocationStrategySpec.scala
new file mode 100644
index 0000000000..d65a35dda2
--- /dev/null
+++
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/ConsistentHashingShardAllocationStrategySpec.scala
@@ -0,0 +1,257 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.cluster.sharding
+
+import scala.collection.immutable.SortedSet
+
+import org.apache.pekko
+import pekko.actor.ActorPath
+import pekko.actor.ActorRef
+import pekko.actor.ActorRefProvider
+import pekko.actor.Address
+import pekko.actor.MinimalActorRef
+import pekko.actor.RootActorPath
+import pekko.cluster.ClusterEvent
+import pekko.cluster.ClusterEvent.CurrentClusterState
+import pekko.cluster.ClusterSettings
+import pekko.cluster.Member
+import pekko.cluster.MemberStatus
+import pekko.cluster.UniqueAddress
+import pekko.event.Logging
+import pekko.event.LoggingAdapter
+import pekko.testkit.PekkoSpec
+import pekko.util.Version
+
+object ConsistentHashingShardAllocationStrategySpec {
+
+ final class DummyActorRef(val path: ActorPath) extends MinimalActorRef {
+ override def provider: ActorRefProvider = ???
+ }
+
+ def newUpMember(host: String, port: Int = 252525, version: Version =
Version("1.0.0")) =
+ Member(
+ UniqueAddress(Address("pekko", "myapp", host, port), 1L),
+ Set(ClusterSettings.DcRolePrefix + ClusterSettings.DefaultDataCenter),
+ version).copy(MemberStatus.Up)
+
+ def newFakeRegion(idForDebug: String, member: Member): ActorRef =
+ new DummyActorRef(RootActorPath(member.address) / "system" / "fake" /
idForDebug)
+}
+
+class ConsistentHashingShardAllocationStrategySpec extends PekkoSpec {
+ import ConsistentHashingShardAllocationStrategySpec._
+
+ val memberA = newUpMember("127.0.0.1")
+ val memberB = newUpMember("127.0.0.2")
+ val memberC = newUpMember("127.0.0.3")
+ val memberD = newUpMember("127.0.0.4")
+
+ val regionA = newFakeRegion("regionA", memberA)
+ val regionB = newFakeRegion("regionB", memberB)
+ val regionC = newFakeRegion("regionC", memberC)
+ val regionD = newFakeRegion("regionD", memberD)
+
+ private val emptyAllocationsABC: Map[ActorRef, Vector[String]] =
+ Map(regionA -> Vector.empty, regionB -> Vector.empty, regionC ->
Vector.empty)
+
+ private def strategy(rebalanceLimit: Int = 0) =
+ // we don't really "start" it as we fake the cluster access
+ new ConsistentHashingShardAllocationStrategy(rebalanceLimit) {
+ override protected def clusterState: ClusterEvent.CurrentClusterState =
+ CurrentClusterState(SortedSet(memberA, memberB, memberC))
+ override protected def selfMember: Member = memberA
+ override protected val log: LoggingAdapter =
+ Logging(system, classOf[ConsistentHashingShardAllocationStrategy])
+ }
+
+ "ConsistentHashingShardAllocationStrategy" must {
+ "allocate to regions" in {
+ val allocationStrategy = strategy()
+ val allocations = emptyAllocationsABC
+ allocationStrategy.allocateShard(regionA, "0", allocations).futureValue
should ===(regionC)
+ allocationStrategy.allocateShard(regionA, "1", allocations).futureValue
should ===(regionB)
+ allocationStrategy.allocateShard(regionA, "2", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "10", allocations).futureValue
should ===(regionA)
+ }
+
+ "allocate to mostly same regions when node is removed" in {
+ val allocationStrategy = strategy()
+ val allocations = emptyAllocationsABC
+ allocationStrategy.allocateShard(regionA, "0", allocations).futureValue
should ===(regionC)
+ allocationStrategy.allocateShard(regionA, "1", allocations).futureValue
should ===(regionB)
+ allocationStrategy.allocateShard(regionA, "2", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "3", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "10", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "14", allocations).futureValue
should ===(regionA)
+
+ val allocations2 = allocations - regionC
+ allocationStrategy.allocateShard(regionA, "0", allocations2).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "1", allocations2).futureValue
should ===(regionB)
+ allocationStrategy.allocateShard(regionA, "2", allocations2).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "3", allocations2).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "10",
allocations2).futureValue should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "14",
allocations2).futureValue should ===(regionA)
+ }
+
+ "allocate to mostly same regions when node is added" in {
+ val allocationStrategy = strategy()
+ val allocations = emptyAllocationsABC
+ allocationStrategy.allocateShard(regionA, "0", allocations).futureValue
should ===(regionC)
+ allocationStrategy.allocateShard(regionA, "1", allocations).futureValue
should ===(regionB)
+ allocationStrategy.allocateShard(regionA, "2", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "3", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "10", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "14", allocations).futureValue
should ===(regionA)
+
+ val allocations2 = allocations.updated(regionD, Vector.empty)
+ allocationStrategy.allocateShard(regionA, "0", allocations2).futureValue
should ===(regionD)
+ allocationStrategy.allocateShard(regionA, "1", allocations2).futureValue
should ===(regionB)
+ allocationStrategy.allocateShard(regionA, "2", allocations2).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "3", allocations2).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "10",
allocations2).futureValue should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "14",
allocations2).futureValue should ===(regionA)
+ }
+
+ "not rebalance when nodes not changed" in {
+ val allocationStrategy = strategy()
+ val allocations = emptyAllocationsABC
+ allocationStrategy.allocateShard(regionA, "0", allocations).futureValue
should ===(regionC)
+ allocationStrategy.allocateShard(regionA, "1", allocations).futureValue
should ===(regionB)
+ allocationStrategy.allocateShard(regionA, "2", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "10", allocations).futureValue
should ===(regionA)
+
+ val allocations2 = Map(regionA -> Vector("2", "10"), regionB ->
Vector("1"), regionC -> Vector("0"))
+ allocationStrategy.rebalance(allocations2, Set.empty).futureValue should
===(Set.empty[String])
+ }
+
+ "rebalance when node is added" in {
+ val allocationStrategy = strategy()
+ val allocations = emptyAllocationsABC
+ allocationStrategy.allocateShard(regionA, "0", allocations).futureValue
should ===(regionC)
+ allocationStrategy.allocateShard(regionA, "1", allocations).futureValue
should ===(regionB)
+ allocationStrategy.allocateShard(regionA, "2", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "3", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "10", allocations).futureValue
should ===(regionA)
+ allocationStrategy.allocateShard(regionA, "14", allocations).futureValue
should ===(regionA)
+
+ val allocations2 = Map(
+ regionA -> Vector("2", "3", "10", "14"),
+ regionB -> Vector("1"),
+ regionC -> Vector("0"),
+ regionD -> Vector.empty)
+ allocationStrategy.rebalance(allocations2, Set.empty).futureValue should
===(Set("0"))
+
+ val allocations3 = Map(
+ regionB -> Vector("1"),
+ regionA -> Vector("2", "3", "10", "14"),
+ regionD -> Vector.empty,
+ regionC -> Vector("0"))
+ allocationStrategy.rebalance(allocations3, Set.empty).futureValue should
===(Set("0"))
+ }
+
+ "not rebalance more than limit" in {
+ val allocationStrategy = strategy(rebalanceLimit = 2)
+ val allocations = Map(
+ regionA -> Vector("0", "1", "2", "3", "10", "14"),
+ regionB -> Vector.empty,
+ regionC -> Vector.empty,
+ regionD -> Vector.empty)
+ allocationStrategy.rebalance(allocations, Set.empty).futureValue should
===(Set("0", "1"))
+
+ val allocations2 = Map(
+ regionA -> Vector("2", "3", "10", "14"),
+ regionB -> Vector("1"),
+ regionC -> Vector("0"),
+ regionD -> Vector.empty)
+ allocationStrategy.rebalance(allocations2, Set.empty).futureValue should
===(Set("0"))
+
+ val allocations3 =
+ Map(regionA -> Vector("2", "3", "10", "14"), regionB -> Vector("1"),
regionC -> Vector.empty,
+ regionD -> Vector("0"))
+ allocationStrategy.rebalance(allocations3, Set.empty).futureValue should
===(Set.empty[String])
+ }
+
+ "not rebalance those that are in progress" in {
+ val allocationStrategy = strategy(rebalanceLimit = 2)
+ val allocations = Map(
+ regionA -> Vector("0", "1", "2", "3", "10", "14"),
+ regionB -> Vector.empty,
+ regionC -> Vector.empty,
+ regionD -> Vector.empty)
+ allocationStrategy.rebalance(allocations, Set.empty).futureValue should
===(Set("0", "1"))
+ allocationStrategy.rebalance(allocations, Set("0", "1")).futureValue
should ===(Set.empty[String])
+ // 2, 3, 10 and 14 are already at right place
+ allocationStrategy.rebalance(allocations, Set("0", "1", "2",
"3")).futureValue should ===(Set.empty[String])
+ }
+
+ "not rebalance when rolling update in progress" in {
+ val allocationStrategy =
+ new ConsistentHashingShardAllocationStrategy(rebalanceLimit = 0) {
+
+ val member1 = newUpMember("127.0.0.1", version = Version("1.0.0"))
+ val member2 = newUpMember("127.0.0.2", version = Version("1.0.1"))
+ val member3 = newUpMember("127.0.0.3", version = Version("1.0.0"))
+
+ // multiple versions to simulate rolling update in progress
+ override protected def clusterState: CurrentClusterState =
+ CurrentClusterState(SortedSet(member1, member2, member3))
+
+ override protected def selfMember: Member = member1
+
+ override protected val log: LoggingAdapter =
+ Logging(system, classOf[ConsistentHashingShardAllocationStrategy])
+ }
+ val allocations = Map(regionA -> Vector("0", "1", "2", "3", "10", "14"),
regionB -> Vector.empty)
+ allocationStrategy.rebalance(allocations, Set.empty).futureValue should
===(Set.empty[String])
+ }
+
+ "not rebalance when regions are unreachable" in {
+ val allocationStrategy =
+ new ConsistentHashingShardAllocationStrategy(rebalanceLimit = 0) {
+
+ override protected def clusterState: CurrentClusterState =
+ CurrentClusterState(SortedSet(memberA, memberB, memberC),
unreachable = Set(memberB))
+ override protected def selfMember: Member = memberB
+ override protected val log: LoggingAdapter =
+ Logging(system, classOf[ConsistentHashingShardAllocationStrategy])
+ }
+ val allocations =
+ Map(regionA -> Vector("0", "1", "2", "3", "10", "14"), regionB ->
Vector.empty, regionC -> Vector.empty)
+ allocationStrategy.rebalance(allocations, Set.empty).futureValue should
===(Set.empty[String])
+ }
+
+ "not rebalance when members are joining dc" in {
+ val allocationStrategy =
+ new ConsistentHashingShardAllocationStrategy(rebalanceLimit = 0) {
+
+ val member1 = newUpMember("127.0.0.1")
+ val member2 =
+ Member(
+ UniqueAddress(Address("pekko", "myapp", "127.0.0.2", 252525),
1L),
+ Set(ClusterSettings.DcRolePrefix +
ClusterSettings.DefaultDataCenter),
+ member1.appVersion)
+ val member3 = newUpMember("127.0.0.3")
+
+ override protected def clusterState: CurrentClusterState =
+ CurrentClusterState(SortedSet(member1, member2, member3),
unreachable = Set.empty)
+ override protected def selfMember: Member = member2
+ override protected val log: LoggingAdapter =
+ Logging(system, classOf[ConsistentHashingShardAllocationStrategy])
+ }
+ val allocations =
+ Map(regionA -> Vector("0", "1", "2", "3", "10", "14"), regionB ->
Vector.empty, regionC -> Vector.empty)
+ allocationStrategy.rebalance(allocations, Set.empty).futureValue should
===(Set.empty[String])
+ }
+ }
+}
diff --git
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/DeprecatedLeastShardAllocationStrategySpec.scala
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/DeprecatedLeastShardAllocationStrategySpec.scala
index 7b1a09c9d6..88e6c3d524 100644
---
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/DeprecatedLeastShardAllocationStrategySpec.scala
+++
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/DeprecatedLeastShardAllocationStrategySpec.scala
@@ -23,8 +23,8 @@ import pekko.cluster.ClusterSettings
import pekko.cluster.Member
import pekko.cluster.MemberStatus
import pekko.cluster.UniqueAddress
-import pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy
-import
pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy.RegionEntry
+import pekko.cluster.sharding.internal.ClusterShardAllocationMixin.RegionEntry
+import
pekko.cluster.sharding.internal.ClusterShardAllocationMixin.ShardSuitabilityOrdering
import pekko.testkit.PekkoSpec
import pekko.util.Version
@@ -202,7 +202,7 @@ class DeprecatedLeastShardAllocationStrategySpec extends
PekkoSpec {
RegionEntry(fakeRegionC, newVersionMember2, Vector("ShardId1")))
val sortedRegions =
-
shardsAndMembers.sorted(AbstractLeastShardAllocationStrategy.ShardSuitabilityOrdering).map(_.region)
+ shardsAndMembers.sorted(ShardSuitabilityOrdering).map(_.region)
// only node b has the new version
sortedRegions should ===(
@@ -219,11 +219,13 @@ class DeprecatedLeastShardAllocationStrategySpec extends
PekkoSpec {
new ShardCoordinator.LeastShardAllocationStrategy(rebalanceThreshold =
2, maxSimultaneousRebalance = 100) {
val member1 = newUpMember("127.0.0.1", version = Version("1.0.0"))
- val member2 = newUpMember("127.0.0.1", version = Version("1.0.1"))
+ val member2 = newUpMember("127.0.0.2", version = Version("1.0.1"))
+ val member3 = newUpMember("127.0.0.3", version = Version("1.0.0"))
// multiple versions to simulate rolling update in progress
override protected def clusterState: CurrentClusterState =
- CurrentClusterState(SortedSet(member1, member2))
+ CurrentClusterState(SortedSet(member1, member2, member3))
+
override protected def selfMember: Member = member1
}
val allocations = createAllocations(aCount = 5, bCount = 5)
@@ -237,13 +239,9 @@ class DeprecatedLeastShardAllocationStrategySpec extends
PekkoSpec {
val allocationStrategy =
new ShardCoordinator.LeastShardAllocationStrategy(rebalanceThreshold =
2, maxSimultaneousRebalance = 100) {
- val member1 = newUpMember("127.0.0.1")
- val member2 = newUpMember("127.0.0.2")
-
- // multiple versions to simulate rolling update in progress
override protected def clusterState: CurrentClusterState =
- CurrentClusterState(SortedSet(member1, member2), unreachable =
Set(member2))
- override protected def selfMember: Member = member1
+ CurrentClusterState(SortedSet(memberA, memberB, memberC),
unreachable = Set(memberB))
+ override protected def selfMember: Member = memberB
}
val allocations = createAllocations(aCount = 5, bCount = 5)
allocationStrategy.rebalance(allocations, Set.empty).futureValue should
===(Set.empty[String])
@@ -262,10 +260,10 @@ class DeprecatedLeastShardAllocationStrategySpec extends
PekkoSpec {
UniqueAddress(Address("pekko", "myapp", "127.0.0.2", 252525),
1L),
Set(ClusterSettings.DcRolePrefix +
ClusterSettings.DefaultDataCenter),
member1.appVersion)
+ val member3 = newUpMember("127.0.0.3")
- // multiple versions to simulate rolling update in progress
override protected def clusterState: CurrentClusterState =
- CurrentClusterState(SortedSet(member1, member2), unreachable =
Set(member2))
+ CurrentClusterState(SortedSet(member1, member2, member3),
unreachable = Set.empty)
override protected def selfMember: Member = member2
}
val allocations = createAllocations(aCount = 5, bCount = 5)
diff --git
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategySpec.scala
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategySpec.scala
index 44da6b3fa4..50a5535ab4 100644
---
a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategySpec.scala
+++
b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategySpec.scala
@@ -31,8 +31,8 @@ import pekko.cluster.MemberStatus
import pekko.cluster.UniqueAddress
import pekko.cluster.sharding.ShardCoordinator.ShardAllocationStrategy
import pekko.cluster.sharding.ShardRegion.ShardId
-import pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy
-import
pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy.RegionEntry
+import pekko.cluster.sharding.internal.ClusterShardAllocationMixin.RegionEntry
+import
pekko.cluster.sharding.internal.ClusterShardAllocationMixin.ShardSuitabilityOrdering
import pekko.cluster.sharding.internal.LeastShardAllocationStrategy
import pekko.testkit.PekkoSpec
import pekko.util.Version
@@ -257,7 +257,7 @@ class LeastShardAllocationStrategySpec extends PekkoSpec {
RegionEntry(fakeRegionC, newVersionMember2, Vector("ShardId1")))
val sortedRegions =
-
shardsAndMembers.sorted(AbstractLeastShardAllocationStrategy.ShardSuitabilityOrdering).map(_.region)
+ shardsAndMembers.sorted(ShardSuitabilityOrdering).map(_.region)
// only node b has the new version
sortedRegions should ===(
@@ -274,11 +274,12 @@ class LeastShardAllocationStrategySpec extends PekkoSpec {
new LeastShardAllocationStrategy(absoluteLimit = 1000, relativeLimit =
1.0) {
val member1 = newUpMember("127.0.0.1", version = Version("1.0.0"))
- val member2 = newUpMember("127.0.0.1", version = Version("1.0.1"))
+ val member2 = newUpMember("127.0.0.2", version = Version("1.0.1"))
+ val member3 = newUpMember("127.0.0.3", version = Version("1.0.0"))
// multiple versions to simulate rolling update in progress
override protected def clusterState: CurrentClusterState =
- CurrentClusterState(SortedSet(member1, member2))
+ CurrentClusterState(SortedSet(member1, member2, member3))
override protected def selfMember: Member = member1
}
@@ -293,13 +294,9 @@ class LeastShardAllocationStrategySpec extends PekkoSpec {
val allocationStrategy =
new LeastShardAllocationStrategy(absoluteLimit = 1000, relativeLimit =
1.0) {
- val member1 = newUpMember("127.0.0.1")
- val member2 = newUpMember("127.0.0.2")
-
- // multiple versions to simulate rolling update in progress
override protected def clusterState: CurrentClusterState =
- CurrentClusterState(SortedSet(member1, member2), unreachable =
Set(member2))
- override protected def selfMember: Member = member2
+ CurrentClusterState(SortedSet(memberA, memberB, memberC),
unreachable = Set(memberB))
+ override protected def selfMember: Member = memberB
}
val allocations = createAllocations(aCount = 5, bCount = 5)
allocationStrategy.rebalance(allocations, Set.empty).futureValue should
===(Set.empty[String])
@@ -317,10 +314,10 @@ class LeastShardAllocationStrategySpec extends PekkoSpec {
UniqueAddress(Address("pekko", "myapp", "127.0.0.2", 252525),
1L),
Set(ClusterSettings.DcRolePrefix +
ClusterSettings.DefaultDataCenter),
member1.appVersion)
+ val member3 = newUpMember("127.0.0.3")
- // multiple versions to simulate rolling update in progress
override protected def clusterState: CurrentClusterState =
- CurrentClusterState(SortedSet(member1, member2), unreachable =
Set(member2))
+ CurrentClusterState(SortedSet(member1, member2, member3),
unreachable = Set.empty)
override protected def selfMember: Member = member2
}
val allocations = createAllocations(aCount = 5, bCount = 5)
diff --git a/docs/src/main/paradox/persistence-query.md
b/docs/src/main/paradox/persistence-query.md
index 44f2de0bc1..a85199a8fd 100644
--- a/docs/src/main/paradox/persistence-query.md
+++ b/docs/src/main/paradox/persistence-query.md
@@ -172,6 +172,11 @@ See
@apidoc[pekko.persistence.query.typed.*.EventsBySliceQuery] and @apidoc[pekk
A variation of these are
@apidoc[pekko.persistence.query.typed.*.EventsBySliceStartingFromSnapshotsQuery]
and
@apidoc[pekko.persistence.query.typed.*.CurrentEventsBySliceStartingFromSnapshotsQuery].
+@apidoc[pekko.persistence.query.typed.*.EventsBySliceFirehoseQuery] can give
better scalability when many
+consumers retrieve the same events, for example many Projections of the same
entity type. The purpose is
+to share the stream of events from the database and fan out to connected
consumer streams. Thereby fewer queries
+and loading of events from the database. It is typically used together with
@ref:[Sharded Daemon Process with colocated
processes](typed/cluster-sharded-daemon-process.md#colocate-processes).
+
### Materialized values of queries
Journals are able to provide additional information related to a query by
exposing @ref:[Materialized
values](stream/stream-quickstart.md#materialized-values-quick),
diff --git a/docs/src/main/paradox/typed/cluster-sharded-daemon-process.md
b/docs/src/main/paradox/typed/cluster-sharded-daemon-process.md
index dcc31f29ce..26c8d6925f 100644
--- a/docs/src/main/paradox/typed/cluster-sharded-daemon-process.md
+++ b/docs/src/main/paradox/typed/cluster-sharded-daemon-process.md
@@ -18,8 +18,8 @@ To use Pekko Sharded Daemon Process, you must add the
following dependency in yo
## Introduction
Sharded Daemon Process provides a way to run `N` actors, each given a numeric
id starting from `0` that are then kept alive
-and balanced across the cluster. When a rebalance is needed the actor is
stopped and, triggered by a keep alive running on
-all nodes, started on a new node (the keep alive should be seen as an
implementation detail and may change in future versions).
+and balanced across the cluster. When a rebalance is needed the actor is
stopped and, triggered by a keep alive from
+a Cluster Singleton (the keep alive should be seen as an implementation detail
and may change in future versions).
The intended use case is for splitting data processing workloads across a set
number of workers that each get to work on a subset
of the data that needs to be processed. This is commonly needed to create
projections based on the event streams available
@@ -46,6 +46,40 @@ An additional factory method is provided for further
configurability and providi
In use cases where you need to send messages to the daemon process actors it
is recommended to use the @ref:[system receptionist](actor-discovery.md)
either with a single `ServiceKey` which all daemon process actors register
themeselves to for broadcasts or individual keys if more fine grained messaging
is needed.
+## Dynamic scaling of number of workers
+
+Starting the sharded daemon process with `initWithContext` returns an
`ActorRef[ShardedDaemonProcessCommand]` that accepts a
@apidoc[ChangeNumberOfProcesses] command to rescale the process to a new number
of workers.
+
+The rescaling process among other things includes the process actors stopping
themselves in response to a stop message
+so may be a relatively slow operation. If a subsequent request to rescale is
sent while one is in progress it is responded
+to with a failure response.
+
+A rolling upgrade switching from a static number of workers to a dynamic
number is possible.
+It is not safe to do a rolling upgrade from dynamic number of workers to
static without a full cluster shutdown.
+
+### Colocate processes
+
+When using the default shard allocation strategy the processes for different
names are allocated independent of
+each other, i.e. the same process index for different process names may be
allocated to different nodes.
+Colocating processes can be useful to share resources, such as Projections
with
@ref:[EventsBySliceFirehoseQuery](../persistence-query.md#eventsbyslice-and-currenteventsbyslice)
+
+To colocate such processes you can use the
@apidoc[pekko.cluster.sharding.ConsistentHashingShardAllocationStrategy]
+as `shardAllocationStrategy` parameter of the `init` or `initWithContext`
methods.
+
+@@@ note
+Create a new instance of the `ConsistentHashingShardAllocationStrategy` for
each Sharded Daemon Process name, i.e. a
`ConsistentHashingShardAllocationStrategy` instance must not be shared.
+@@@
+
+The shard identifier that is used by Sharded Daemon Process is the same as the
process index, i.e. processes with
+the same index will be colocated.
+
+The allocation strategy is using [Consistent
Hashing](https://tom-e-white.com/2007/11/consistent-hashing.html)
+of the Cluster membership ring to assign a shard to a node. When adding or
removing nodes it will rebalance
+according to the new consistent hashing, but that means that only a few shards
will be rebalanced and others
+remain on the same location. When there are changes to the Cluster membership
the shards may be on different
+nodes for a while, but eventually, when the membership is stable, the shards
with the same identifier will
+end up on the same node.
+
## Scalability
This cluster tool is intended for small numbers of consumers and will not
scale well to a large set. In large clusters
diff --git a/docs/src/main/paradox/typed/cluster-sharding.md
b/docs/src/main/paradox/typed/cluster-sharding.md
index 1192a2ce5e..25be120b96 100644
--- a/docs/src/main/paradox/typed/cluster-sharding.md
+++ b/docs/src/main/paradox/typed/cluster-sharding.md
@@ -236,6 +236,44 @@ support a greater number of shards.
is an example project that can be downloaded, and with instructions of how to
run, that demonstrates how to use
external sharding to co-locate Kafka partition consumption with shards.
+### Colocate Shards
+
+When using the default shard allocation strategy the shards for different
entity types are allocated independent of
+each other, i.e. the same shard identifier for the different entity types may
be allocated to different nodes.
+Colocating shards can be useful if it's known that certain entities interact
or share resources with some other
+entities and that can be defined by using the same shard identifier.
+
+To colocate such shards you can use the
@apidoc[pekko.cluster.sharding.ConsistentHashingShardAllocationStrategy].
+
+Let's look at an example where the purpose is to colocate `Device` entities
with the `Building` entity they belong to.
+To use the same shard identifier we need to use a custom
@apidoc[pekko.cluster.sharding.typed.ShardingMessageExtractor]
+for the `Device` and `Building` entities:
+
+Scala
+: @@snip
[ConsistentHashingShardAllocationCompileOnlySpec](/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlySpec.scala)
{ #building #device }
+
+Java
+: @@snip
[ConsistentHashingShardAllocationCompileOnlyTest](/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlyTest.java)
{ #building #device }
+
+Set the allocation strategy and message extractor on your
@apidoc[typed.*.Entity]:
+
+Scala
+: @@snip
[ConsistentHashingShardAllocationCompileOnlySpec](/cluster-sharding-typed/src/test/scala/docs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlySpec.scala)
{ #init }
+
+Java
+: @@snip
[ConsistentHashingShardAllocationCompileOnlyTest](/cluster-sharding-typed/src/test/java/jdocs/org/apache/pekko/cluster/sharding/typed/ConsistentHashingShardAllocationCompileOnlyTest.java)
{ #init }
+
+@@@ note
+Create a new instance of the `ConsistentHashingShardAllocationStrategy` for
each entity type, i.e. a `ConsistentHashingShardAllocationStrategy` instance
must not be shared between different entity types.
+@@@
+
+The allocation strategy is using [Consistent
Hashing](https://tom-e-white.com/2007/11/consistent-hashing.html)
+of the Cluster membership ring to assign a shard to a node. When adding or
removing nodes it will rebalance
+according to the new consistent hashing, but that means that only a few shards
will be rebalanced and others
+remain on the same location. When there are changes to the Cluster membership
the shards may be on different
+nodes for a while, but eventually, when the membership is stable, the shards
with the same identifier will
+end up on the same node.
+
### Custom shard allocation
An optional custom shard allocation strategy can be passed into the optional
parameter when initializing an entity type
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
