Merge pull request #82 from lerna-stack/improve-raft-actor-allocation

✨feat: Distribute the shards to nodes fairly

Author: tksugimoto

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 [Unreleased]: https://github.com/lerna-stack/akka-entity-replication/compare/.v1.0.0...master
 
+### Breaking Change
+
+- Change the shard-distribution-strategy to distribute shard (`RaftActor`) more evenly [PR#82](https://github.com/lerna-stack/akka-entity-replication/pull/82)
+
+  ⚠️ This change does not allow rolling updates. You have to update your system by stopping the whole cluster.
+
 ### Added
 - Java11 support
 - Add new typed API based on Akka Typed [PR#79](https://github.com/lerna-stack/akka-entity-replication/pull/79)

README.md

@@ -29,7 +29,9 @@ To use this library, you must add a dependency into your sbt project, add the fo
 
 **Stable Release**
 
-[![Maven Central](https://img.shields.io/maven-central/v/com.lerna-stack/akka-entity-replication_2.13?color=%23005cb2&label=stable)](https://mvnrepository.com/artifact/com.lerna-stack/akka-entity-replication)
+[![Maven Central](https://img.shields.io/maven-central/v/com.lerna-stack/akka-entity-replication_2.13?color=%23005cb2&label=stable)](https://mvnrepository.com/artifact/com.lerna-stack/akka-entity-replication) 
+
+⚠️ `v2.0.0` will contain breaking changes. For more details see [CHANGELOG](./CHANGELOG.md).
 
 ```scala
 libraryDependencies += "com.lerna-stack" %% "akka-entity-replication" % "X.X.X"

src/main/scala/lerna/akka/entityreplication/ReplicationRegion.scala

@@ -2,7 +2,7 @@ package lerna.akka.entityreplication
 
 import akka.actor.{ Actor, ActorLogging, ActorPath, ActorRef, OneForOneStrategy, Props, Stash, SupervisorStrategy }
 import akka.cluster.ClusterEvent._
-import akka.cluster.sharding.ShardRegion.{ GracefulShutdown, HashCodeMessageExtractor }
+import akka.cluster.sharding.ShardRegion.GracefulShutdown
 import akka.cluster.sharding.{ ClusterSharding, ClusterShardingSettings, ShardRegion }
 import akka.cluster.{ Cluster, Member, MemberStatus }
 import akka.routing.{ ActorRefRoutee, ConsistentHashingRouter, ConsistentHashingRoutingLogic, Router }
@@ -118,15 +118,20 @@ private[entityreplication] class ReplicationRegion(
 
   // TODO 変数名を実態にあったものに変更
   private[this] val shardingRouters: Map[MemberIndex, ActorRef] = allMemberIndexes.map { memberIndex =>
+    def clusterReplicationShardId(message: Any): String = extractNormalizedShardIdInternal(message).raw
+    val extractEntityId: ShardRegion.ExtractEntityId    = message => (clusterReplicationShardId(message), message)
+    val extractShardId: ShardRegion.ExtractShardId = {
+      case ShardRegion.StartEntity(id) => id
+      case message                     => clusterReplicationShardId(message)
+    }
     memberIndex -> {
       ClusterSharding(context.system).start(
         typeName = s"raft-shard-$typeName-${memberIndex.role}",
         entityProps = createRaftActorProps(),
         settings = ClusterShardingSettings(settings.raftSettings.clusterShardingConfig)
           .withRole(memberIndex.role),
-        messageExtractor = new HashCodeMessageExtractor(maxNumberOfShards = 50) {
-          override def entityId(message: Any): String = extractNormalizedShardIdInternal(message).raw
-        },
+        extractEntityId,
+        extractShardId,
       )
     }
   }.toMap