raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Clear out list of shards in DData coordinator store when no longer needed (#29232)

Browse source
This commit is contained in:
Johan Andrén 2020-06-15 17:36:22 +02:00 • committed by GitHub
parent b70d851ea0
commit 988ead1ef6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 13 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

13
akka-cluster-sharding/src/main/scala/akka/cluster/sharding/internal/DDataRememberEntitiesCoordinatorStore.scala
View file

@ -58,7 +58,9 @@ private[akka] final class DDataRememberEntitiesCoordinatorStore(
override def receive: Receive = { override def receive: Receive = {
case RememberEntitiesCoordinatorStore.GetShards => case RememberEntitiesCoordinatorStore.GetShards =>
allShards match { allShards match {
case Some(shardIds) => sender() ! RememberEntitiesCoordinatorStore.RememberedShards(shardIds) case Some(shardIds) =>
coordinatorWaitingForShards = Some(sender())
onGotAllShards(shardIds);
case None => case None =>
// reply when we get them, since there is only ever one coordinator communicating with us // reply when we get them, since there is only ever one coordinator communicating with us
// and it may retry we can just keep the latest sender // and it may retry we can just keep the latest sender
@ -104,10 +106,15 @@ private[akka] final class DDataRememberEntitiesCoordinatorStore(
} }
def onGotAllShards(shardIds: Set[ShardId]): Unit = { def onGotAllShards(shardIds: Set[ShardId]): Unit = {
allShards = Some(shardIds) coordinatorWaitingForShards match {
coordinatorWaitingForShards.foreach { coordinator => case Some(coordinator) =>
coordinator ! RememberEntitiesCoordinatorStore.RememberedShards(shardIds) coordinator ! RememberEntitiesCoordinatorStore.RememberedShards(shardIds)
coordinatorWaitingForShards = None coordinatorWaitingForShards = None
// clear the shards out now that we have sent them to coordinator, to save some memory
allShards = None
case None =>
// wait for coordinator to ask
allShards = Some(shardIds)
} }
} }