parent
22f31b4d69
commit
5fce4bea63
5 changed files with 78 additions and 49 deletions
|
|
@ -24,8 +24,8 @@ akka.cluster.sharding {
|
||||||
remember-entities = off
|
remember-entities = off
|
||||||
|
|
||||||
# Set this to a time duration to have sharding passivate entities when they have not
|
# Set this to a time duration to have sharding passivate entities when they have not
|
||||||
# gotten any message in this long time. Set to 'off' to disable.
|
# received any message in this length of time. Set to 'off' to disable.
|
||||||
passivate-idle-entity-after = off
|
passivate-idle-entity-after = 120s
|
||||||
|
|
||||||
# If the coordinator can't store state changes it will be stopped
|
# If the coordinator can't store state changes it will be stopped
|
||||||
# and started again after this duration, with an exponential back-off
|
# and started again after this duration, with an exponential back-off
|
||||||
|
|
|
||||||
|
|
@ -4,24 +4,32 @@
|
||||||
|
|
||||||
package akka.cluster.sharding
|
package akka.cluster.sharding
|
||||||
|
|
||||||
|
import scala.concurrent.duration._
|
||||||
|
|
||||||
import akka.actor.{ Actor, ActorRef, Props }
|
import akka.actor.{ Actor, ActorRef, Props }
|
||||||
import akka.cluster.Cluster
|
import akka.cluster.Cluster
|
||||||
import akka.cluster.sharding.InactiveEntityPassivationSpec.Entity.GotIt
|
import akka.cluster.sharding.InactiveEntityPassivationSpec.Entity.GotIt
|
||||||
import akka.testkit.{ AkkaSpec, TestProbe }
|
import akka.testkit.{ AkkaSpec, TestProbe }
|
||||||
import com.typesafe.config.ConfigFactory
|
import com.typesafe.config.ConfigFactory
|
||||||
|
import com.typesafe.config.Config
|
||||||
import scala.concurrent.duration._
|
|
||||||
|
|
||||||
object InactiveEntityPassivationSpec {
|
object InactiveEntityPassivationSpec {
|
||||||
|
|
||||||
val config = ConfigFactory.parseString("""
|
val config = ConfigFactory.parseString("""
|
||||||
akka.loglevel = INFO
|
akka.loglevel = INFO
|
||||||
akka.actor.provider = "cluster"
|
akka.actor.provider = "cluster"
|
||||||
akka.remote.netty.tcp.port = 0
|
akka.remote.netty.tcp.port = 0
|
||||||
akka.remote.artery.canonical.port = 0
|
akka.remote.artery.canonical.port = 0
|
||||||
akka.cluster.sharding.passivate-idle-entity-after = 3 s
|
|
||||||
akka.actor.serialize-messages = off
|
akka.actor.serialize-messages = off
|
||||||
""")
|
""")
|
||||||
|
|
||||||
|
val enabledConfig = ConfigFactory.parseString("""
|
||||||
|
akka.cluster.sharding.passivate-idle-entity-after = 3 s
|
||||||
|
""").withFallback(config)
|
||||||
|
|
||||||
|
val disabledConfig =
|
||||||
|
ConfigFactory.parseString("""akka.cluster.sharding.passivate-idle-entity-after = off""").withFallback(config)
|
||||||
|
|
||||||
object Passivate
|
object Passivate
|
||||||
object Entity {
|
object Entity {
|
||||||
def props(probe: ActorRef) = Props(new Entity(probe))
|
def props(probe: ActorRef) = Props(new Entity(probe))
|
||||||
|
|
@ -33,11 +41,9 @@ object InactiveEntityPassivationSpec {
|
||||||
|
|
||||||
def receive = {
|
def receive = {
|
||||||
case Passivate =>
|
case Passivate =>
|
||||||
probe ! id + " passivating"
|
|
||||||
context.stop(self)
|
context.stop(self)
|
||||||
case msg => probe ! GotIt(id, msg, System.nanoTime())
|
case msg => probe ! GotIt(id, msg, System.nanoTime())
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
val extractEntityId: ShardRegion.ExtractEntityId = {
|
val extractEntityId: ShardRegion.ExtractEntityId = {
|
||||||
|
|
@ -47,22 +53,19 @@ object InactiveEntityPassivationSpec {
|
||||||
val extractShardId: ShardRegion.ExtractShardId = {
|
val extractShardId: ShardRegion.ExtractShardId = {
|
||||||
case msg: Int => (msg % 10).toString
|
case msg: Int => (msg % 10).toString
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
class InactiveEntityPassivationSpec extends AkkaSpec(InactiveEntityPassivationSpec.config) {
|
abstract class AbstractInactiveEntityPassivationSpec(c: Config) extends AkkaSpec(c) {
|
||||||
import InactiveEntityPassivationSpec._
|
import InactiveEntityPassivationSpec._
|
||||||
|
|
||||||
val smallTolerance = 300.millis
|
private val smallTolerance = 300.millis
|
||||||
|
|
||||||
"Passivation of inactive entities" must {
|
private val settings = ClusterShardingSettings(system)
|
||||||
|
|
||||||
"passivate entities when they haven't seen messages for the configured duration" in {
|
def start(probe: TestProbe): ActorRef = {
|
||||||
// single node cluster
|
// single node cluster
|
||||||
Cluster(system).join(Cluster(system).selfAddress)
|
Cluster(system).join(Cluster(system).selfAddress)
|
||||||
val probe = TestProbe()
|
ClusterSharding(system).start(
|
||||||
val settings = ClusterShardingSettings(system)
|
|
||||||
val region = ClusterSharding(system).start(
|
|
||||||
"myType",
|
"myType",
|
||||||
InactiveEntityPassivationSpec.Entity.props(probe.ref),
|
InactiveEntityPassivationSpec.Entity.props(probe.ref),
|
||||||
settings,
|
settings,
|
||||||
|
|
@ -70,7 +73,9 @@ class InactiveEntityPassivationSpec extends AkkaSpec(InactiveEntityPassivationSp
|
||||||
extractShardId,
|
extractShardId,
|
||||||
ClusterSharding(system).defaultShardAllocationStrategy(settings),
|
ClusterSharding(system).defaultShardAllocationStrategy(settings),
|
||||||
Passivate)
|
Passivate)
|
||||||
|
}
|
||||||
|
|
||||||
|
def timeUntilPassivate(region: ActorRef, probe: TestProbe): FiniteDuration = {
|
||||||
region ! 1
|
region ! 1
|
||||||
region ! 2
|
region ! 2
|
||||||
val responses = Set(probe.expectMsgType[GotIt], probe.expectMsgType[GotIt])
|
val responses = Set(probe.expectMsgType[GotIt], probe.expectMsgType[GotIt])
|
||||||
|
|
@ -83,18 +88,36 @@ class InactiveEntityPassivationSpec extends AkkaSpec(InactiveEntityPassivationSp
|
||||||
region ! 2
|
region ! 2
|
||||||
probe.expectMsgType[GotIt].id should ===("2")
|
probe.expectMsgType[GotIt].id should ===("2")
|
||||||
|
|
||||||
// make sure "1" hasn't seen a message in 3 seconds and passivates
|
|
||||||
val timeSinceOneSawAMessage = (System.nanoTime() - timeOneSawMessage).nanos
|
val timeSinceOneSawAMessage = (System.nanoTime() - timeOneSawMessage).nanos
|
||||||
val timeUntilPassivate: FiniteDuration = (3.seconds - timeSinceOneSawAMessage) - smallTolerance
|
(settings.passivateIdleEntityAfter - timeSinceOneSawAMessage) + smallTolerance
|
||||||
probe.expectNoMessage(timeUntilPassivate)
|
}
|
||||||
probe.expectMsg("1 passivating")
|
}
|
||||||
|
|
||||||
// but it can be re activated just fine:
|
class InactiveEntityPassivationSpec
|
||||||
|
extends AbstractInactiveEntityPassivationSpec(InactiveEntityPassivationSpec.enabledConfig) {
|
||||||
|
"Passivation of inactive entities" must {
|
||||||
|
"passivate entities when they haven't seen messages for the configured duration" in {
|
||||||
|
val probe = TestProbe()
|
||||||
|
val region = start(probe)
|
||||||
|
|
||||||
|
// make sure "1" hasn't seen a message in 3 seconds and passivates
|
||||||
|
probe.expectNoMessage(timeUntilPassivate(region, probe))
|
||||||
|
|
||||||
|
// but it can be re activated
|
||||||
region ! 1
|
region ! 1
|
||||||
region ! 2
|
region ! 2
|
||||||
Set(probe.expectMsgType[GotIt], probe.expectMsgType[GotIt]).map(_.id) should ===(Set("1", "2"))
|
Set(probe.expectMsgType[GotIt], probe.expectMsgType[GotIt]).map(_.id) should ===(Set("1", "2"))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
class DisabledInactiveEntityPassivationSpec
|
||||||
|
extends AbstractInactiveEntityPassivationSpec(InactiveEntityPassivationSpec.disabledConfig) {
|
||||||
|
"Passivation of inactive entities" must {
|
||||||
|
"not passivate when passivation is disabled" in {
|
||||||
|
val probe = TestProbe()
|
||||||
|
val region = start(probe)
|
||||||
|
probe.expectNoMessage(timeUntilPassivate(region, probe))
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -344,12 +344,12 @@ are thereafter delivered to a new incarnation of the entity.
|
||||||
|
|
||||||
### Automatic Passivation
|
### Automatic Passivation
|
||||||
|
|
||||||
The entities can be configured to be automatically passivated if they haven't received
|
The entities are automatically passivated if they haven't received a message within the duration configured in
|
||||||
a message for a while using the `akka.cluster.sharding.passivate-idle-entity-after` setting,
|
`akka.cluster.sharding.passivate-idle-entity-after`
|
||||||
or by explicitly setting `ClusterShardingSettings.passivateIdleEntityAfter` to a suitable
|
or by explicitly setting `ClusterShardingSettings.passivateIdleEntityAfter` to a suitable
|
||||||
time to keep the actor alive. Note that only messages sent through sharding are counted, so direct messages
|
time to keep the actor alive. Note that only messages sent through sharding are counted, so direct messages
|
||||||
to the `ActorRef` of the actor or messages that it sends to itself are not counted as activity.
|
to the `ActorRef` or messages that the actor sends to itself are not counted in this activity.
|
||||||
By default automatic passivation is disabled.
|
Passivation can be disabled by setting `akka.cluster.sharding.passivate-idle-entity-after = off`.
|
||||||
|
|
||||||
<a id="cluster-sharding-remembering"></a>
|
<a id="cluster-sharding-remembering"></a>
|
||||||
## Remembering Entities
|
## Remembering Entities
|
||||||
|
|
|
||||||
|
|
@ -24,3 +24,9 @@ If you are still using Scala 2.11 then you must upgrade to 2.12 or 2.13
|
||||||
Actor DSL is a rarely used feature and has been deprecated since `2.5.0`.
|
Actor DSL is a rarely used feature and has been deprecated since `2.5.0`.
|
||||||
Use plain `system.actorOf` instead of the DSL to create Actors if you have been using it.
|
Use plain `system.actorOf` instead of the DSL to create Actors if you have been using it.
|
||||||
|
|
||||||
|
## Cluster Sharding
|
||||||
|
|
||||||
|
### Passivate idle entity
|
||||||
|
The configuration `akka.cluster.sharding.passivate-idle-entity-after` is now enabled by default.
|
||||||
|
Sharding will passivate entities when they have not received any messages after this duration.
|
||||||
|
Set to `off` to disable.
|
||||||
|
|
@ -123,9 +123,9 @@ message if the entity needs to perform some asynchronous cleanup or interactions
|
||||||
|
|
||||||
### Automatic Passivation
|
### Automatic Passivation
|
||||||
|
|
||||||
The entities can be configured to be automatically passivated if they haven't received
|
The entities are automatically passivated if they haven't received a message within the duration configured in
|
||||||
a message for a while using the `akka.cluster.sharding.passivate-idle-entity-after` setting,
|
`akka.cluster.sharding.passivate-idle-entity-after`
|
||||||
or by explicitly setting `ClusterShardingSettings.passivateIdleEntityAfter` to a suitable
|
or by explicitly setting `ClusterShardingSettings.passivateIdleEntityAfter` to a suitable
|
||||||
time to keep the actor alive. Note that only messages sent through sharding are counted, so direct messages
|
time to keep the actor alive. Note that only messages sent through sharding are counted, so direct messages
|
||||||
to the `ActorRef` of the actor or messages that it sends to itself are not counted as activity.
|
to the `ActorRef` or messages that the actor sends to itself are not counted in this activity.
|
||||||
By default automatic passivation is disabled.
|
Passivation can be disabled by setting `akka.cluster.sharding.passivate-idle-entity-after = off`.
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue