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

Add common multi jvm config to cluster tests (#23974)

Browse source
This commit is contained in:
Christopher Batey 2017-12-04 14:23:55 +00:00 • committed by Arnout Engelen
parent bdc740bf0a
commit 76b2cfa676
3 changed files with 13 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

12
akka-cluster-sharding/src/main/scala/akka/cluster/sharding/ClusterSharding.scala
View file

@ -67,15 +67,15 @@ import scala.collection.JavaConverters._
* tagged with a specific role. The `ShardRegion` is created with two application specific
* functions to extract the entity identifier and the shard identifier from incoming messages.
* A shard is a group of entities that will be managed together. For the first message in a
* specific shard the `ShardRegion` request the location of the shard from a central coordinator,
* the [[ShardCoordinator]]. The `ShardCoordinator` decides which `ShardRegion` that
* specific shard the `ShardRegion` requests the location of the shard from a central coordinator,
* the [[ShardCoordinator]]. The `ShardCoordinator` decides which `ShardRegion`
* owns the shard. The `ShardRegion` receives the decided home of the shard
* and if that is the `ShardRegion` instance itself it will create a local child
* actor representing the entity and direct all messages for that entity to it.
* If the shard home is another `ShardRegion` instance messages will be forwarded
* to that `ShardRegion` instance instead. While resolving the location of a
* shard incoming messages for that shard are buffered and later delivered when the
* shard home is known. Subsequent messages to the resolved shard can be delivered
* shard location is known. Subsequent messages to the resolved shard can be delivered
* to the target destination immediately without involving the `ShardCoordinator`.
*
* To make sure that at most one instance of a specific entity actor is running somewhere
@ -106,7 +106,7 @@ import scala.collection.JavaConverters._
* persistent (durable), e.g. with `akka-persistence`, so that it can be recovered at the new
* location.
*
* The logic that decides which shards to rebalance is defined in a pluggable shard
* The logic that decides which shards to rebalance is defined in a plugable shard
* allocation strategy. The default implementation [[ShardCoordinator.LeastShardAllocationStrategy]]
* picks shards for handoff from the `ShardRegion` with most number of previously allocated shards.
* They will then be allocated to the `ShardRegion` with least number of previously allocated shards,
@ -191,7 +191,7 @@ class ClusterSharding(system: ExtendedActorSystem) extends Extension {
/**
* Scala API: Register a named entity type by defining the [[akka.actor.Props]] of the entity actor
* and functions to extract entity and shard identifier from messages. The [[ShardRegion]] actor
* for this type can later be retrieved with the [[#shardRegion]] method.
* for this type can later be retrieved with the [[shardRegion]] method.
*
* Some settings can be configured as described in the `akka.cluster.sharding` section
* of the `reference.conf`.
@ -231,7 +231,7 @@ class ClusterSharding(system: ExtendedActorSystem) extends Extension {
/**
* Register a named entity type by defining the [[akka.actor.Props]] of the entity actor and
* functions to extract entity and shard identifier from messages. The [[ShardRegion]] actor
* for this type can later be retrieved with the [[#shardRegion]] method.
* for this type can later be retrieved with the [[shardRegion]] method.
*
* The default shard allocation strategy [[ShardCoordinator.LeastShardAllocationStrategy]]
* is used. [[akka.actor.PoisonPill]] is used as `handOffStopMessage`.

14
akka-cluster-sharding/src/multi-jvm/scala/akka/cluster/sharding/ClusterShardingCustomShardAllocationSpec.scala
View file

@ -4,14 +4,10 @@
package akka.cluster.sharding
import scala.collection.immutable
import java.io.File
import akka.cluster.sharding.ShardRegion.Passivate
import scala.concurrent.duration._
import org.apache.commons.io.FileUtils
import com.typesafe.config.ConfigFactory
import akka.actor._
import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import akka.cluster.{ Cluster, MultiNodeClusterSpec }
import akka.persistence.Persistence
import akka.persistence.journal.leveldb.SharedLeveldbJournal
import akka.persistence.journal.leveldb.SharedLeveldbStore
@ -19,9 +15,9 @@ import akka.remote.testconductor.RoleName
import akka.remote.testkit.MultiNodeConfig
import akka.remote.testkit.MultiNodeSpec
import akka.remote.testkit.STMultiNodeSpec
import akka.remote.transport.ThrottlerTransportAdapter.Direction
import akka.testkit._
import akka.cluster.sharding.ShardCoordinator.ShardAllocationStrategy
import scala.concurrent.Future
import akka.util.Timeout
import akka.pattern.ask
@ -37,7 +33,7 @@ object ClusterShardingCustomShardAllocationSpec {
case id: Int ⇒ (id.toString, id)
}
val extractShardId: ShardRegion.ExtractShardId = msg ⇒ msg match {
val extractShardId: ShardRegion.ExtractShardId = {
case id: Int ⇒ id.toString
}
@ -98,7 +94,7 @@ abstract class ClusterShardingCustomShardAllocationSpecConfig(val mode: String)
akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"
akka.persistence.snapshot-store.local.dir = "target/ClusterShardingCustomShardAllocationSpec/snapshots"
akka.cluster.sharding.state-store-mode = "$mode"
"""))
""").withFallback(MultiNodeClusterSpec.clusterConfig))
}
object PersistentClusterShardingCustomShardAllocationSpecConfig extends ClusterShardingCustomShardAllocationSpecConfig("persistence")
@ -153,7 +149,7 @@ abstract class ClusterShardingCustomShardAllocationSpec(config: ClusterShardingC
runOn(first) {
system.actorOf(Props[SharedLeveldbStore], "store")
}
enterBarrier("peristence-started")
enterBarrier("persistence-started")
runOn(first, second) {
system.actorSelection(node(first) / "user" / "store") ! Identify(None)

7
akka-cluster-sharding/src/multi-jvm/scala/akka/cluster/sharding/ClusterShardingLeavingSpec.scala
View file

@ -6,15 +6,12 @@ package akka.cluster.sharding
import java.io.File
import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorIdentity
import akka.actor.ActorRef
import akka.actor.Identify
import akka.actor.Props
import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import akka.cluster.MemberStatus
import akka.cluster.{ Cluster, MemberStatus, MultiNodeClusterSpec }
import akka.persistence.Persistence
import akka.persistence.journal.leveldb.SharedLeveldbJournal
import akka.persistence.journal.leveldb.SharedLeveldbStore
@ -81,7 +78,7 @@ abstract class ClusterShardingLeavingSpecConfig(val mode: String) extends MultiN
dir = target/ClusterShardingLeavingSpec/sharding-ddata
map-size = 10 MiB
}
"""))
""").withFallback(MultiNodeClusterSpec.clusterConfig))
}
object PersistentClusterShardingLeavingSpecConfig extends ClusterShardingLeavingSpecConfig("persistence")