0
# Akka Multi-Node TestKit
1
2
The Akka Multi-Node TestKit provides a comprehensive framework for testing distributed Akka applications across multiple JVMs and potentially multiple machines. It enables coordinated testing with barrier synchronization, network failure injection, and distributed system testing capabilities, making it invaluable for testing distributed systems built with Akka.
3
4
## Package Information
5
6
- **Package Name**: akka-multi-node-testkit
7
- **Package Type**: maven
8
- **Language**: Scala
9
- **Coordinates**: `com.typesafe.akka:akka-multi-node-testkit_2.12:2.8.8`
10
- **Installation**: Add to build.sbt: `"com.typesafe.akka" %% "akka-multi-node-testkit" % "2.8.8" % Test`
11
12
## Core Imports
13
14
```scala
15
import akka.remote.testkit.{MultiNodeConfig, MultiNodeSpec, Direction}
16
import akka.remote.testconductor.{TestConductor, RoleName}
17
```
18
19
## Basic Usage
20
21
```scala
22
import akka.remote.testkit.{MultiNodeConfig, MultiNodeSpec}
23
import akka.remote.testconductor.RoleName
24
import com.typesafe.config.ConfigFactory
25
26
// Define test configuration
27
object SampleMultiNodeConfig extends MultiNodeConfig {
28
val first = role("first")
29
val second = role("second")
30
val third = role("third")
31
32
commonConfig(ConfigFactory.parseString("""
33
akka.actor.provider = remote
34
akka.remote.artery.canonical.port = 0
35
"""))
36
}
37
38
// Create multi-node test
39
class SampleMultiNodeSpec extends MultiNodeSpec(SampleMultiNodeConfig)
40
with AnyWordSpecLike with Matchers {
41
42
import SampleMultiNodeConfig._
43
44
def initialParticipants = roles.size
45
46
"A multi-node cluster" must {
47
"wait for all nodes to start" in {
48
enterBarrier("startup")
49
}
50
51
"allow nodes to communicate" in {
52
runOn(first) {
53
// Code that runs only on first node
54
val actor = system.actorOf(Props[SampleActor](), "sample")
55
enterBarrier("actor-created")
56
}
57
58
runOn(second, third) {
59
enterBarrier("actor-created")
60
val actorSelection = system.actorSelection(node(first) / "user" / "sample")
61
// Test communication with first node's actor
62
}
63
64
enterBarrier("test-complete")
65
}
66
}
67
}
68
```
69
70
## Architecture
71
72
The Akka Multi-Node TestKit is built around several key components:
73
74
- **MultiNodeConfig**: Configuration class for defining test participants, roles, and settings
75
- **MultiNodeSpec**: Main test specification class extending TestKit for multi-node coordination
76
- **TestConductor**: Extension providing barrier coordination and network failure injection
77
- **Conductor/Player Pattern**: Controller orchestrates tests while players participate in coordination
78
- **Barrier Synchronization**: Named barriers allow synchronized test execution across nodes
79
- **Network Failure Injection**: Simulate network conditions like disconnections and throttling
80
81
## Capabilities
82
83
### Test Configuration
84
85
Core configuration system for defining multi-node test participants, roles, and Akka settings. Essential for setting up distributed test scenarios.
86
87
```scala { .api }
88
abstract class MultiNodeConfig {
89
def commonConfig(config: Config): Unit
90
def nodeConfig(roles: RoleName*)(configs: Config*): Unit
91
def role(name: String): RoleName
92
def debugConfig(on: Boolean): Config
93
def deployOn(role: RoleName, deployment: String): Unit
94
def deployOnAll(deployment: String): Unit
95
def testTransport(on: Boolean): Unit
96
}
97
```
98
99
[Test Configuration](./test-configuration.md)
100
101
### Multi-Node Test Specification
102
103
Main test framework for creating and executing coordinated multi-node tests with barrier synchronization and node-specific execution.
104
105
```scala { .api }
106
abstract class MultiNodeSpec(
107
myself: RoleName,
108
system: ActorSystem,
109
roles: immutable.Seq[RoleName],
110
deployments: RoleName => Seq[String]
111
) extends TestKit(system) {
112
def initialParticipants: Int
113
def runOn(nodes: RoleName*)(thunk: => Unit): Unit
114
def isNode(nodes: RoleName*): Boolean
115
def enterBarrier(name: String*): Unit
116
def enterBarrier(max: FiniteDuration, name: String*): Unit
117
def node(role: RoleName): ActorPath
118
}
119
```
120
121
[Multi-Node Test Specification](./multi-node-spec.md)
122
123
### Test Conductor
124
125
Extension providing advanced coordination capabilities including barrier management and network failure injection for testing distributed system resilience.
126
127
```scala { .api }
128
class TestConductorExt(system: ExtendedActorSystem)
129
extends Extension with Conductor with Player {
130
131
// Conductor capabilities
132
def startController(participants: Int, name: RoleName, controllerPort: InetSocketAddress): Future[InetSocketAddress]
133
def blackhole(node: RoleName, target: RoleName, direction: Direction): Future[Done]
134
def passThrough(node: RoleName, target: RoleName, direction: Direction): Future[Done]
135
def disconnect(node: RoleName, target: RoleName): Future[Done]
136
def abort(node: RoleName, target: RoleName): Future[Done]
137
def shutdown(node: RoleName, abort: Boolean): Future[Done]
138
def removeNode(node: RoleName): Future[Done]
139
def getNodes: Future[Iterable[RoleName]]
140
141
// Player capabilities
142
def startClient(name: RoleName, controllerAddr: InetSocketAddress): Future[Done]
143
def enter(timeout: Timeout, name: immutable.Seq[String]): Unit
144
def getAddressFor(name: RoleName): Future[Address]
145
}
146
```
147
148
[Test Conductor](./test-conductor.md)
149
150
### System Properties Configuration
151
152
System property based configuration for multi-node test execution including node identification, networking, and coordinator settings.
153
154
```scala { .api }
155
object MultiNodeSpec {
156
val maxNodes: Int
157
val selfName: String
158
val tcpPort: Int
159
val udpPort: Option[Int]
160
val selfPort: Int
161
val serverName: String
162
val serverPort: Int
163
val selfIndex: Int
164
165
def configureNextPortIfFixed(config: Config): Config
166
}
167
```
168
169
[System Properties Configuration](./system-properties.md)
170
171
## Core Types
172
173
```scala { .api }
174
// Role identification
175
case class RoleName(name: String)
176
177
// Network traffic direction for failure injection
178
sealed trait Direction {
179
def includes(other: Direction): Boolean
180
}
181
182
object Direction {
183
case object Send extends Direction {
184
def includes(other: Direction): Boolean = other == Send
185
def getInstance: Direction = this // Java API
186
}
187
case object Receive extends Direction {
188
def includes(other: Direction): Boolean = other == Receive
189
def getInstance: Direction = this // Java API
190
}
191
case object Both extends Direction {
192
def includes(other: Direction): Boolean = true
193
def getInstance: Direction = this // Java API
194
}
195
196
// Java API factory methods
197
def sendDirection(): Direction = Send
198
def receiveDirection(): Direction = Receive
199
def bothDirection(): Direction = Both
200
}
201
202
// Completion marker for operations
203
sealed trait Done
204
case object Done extends Done
205
206
// Helper for awaiting futures with remaining duration
207
class AwaitHelper[T](w: Awaitable[T]) {
208
def await: T
209
}
210
```
211
212
## Essential Configuration
213
214
### System Properties
215
216
Multi-node tests require specific system properties:
217
218
```properties
219
-Dmultinode.max-nodes=3 # Number of participating nodes
220
-Dmultinode.host=localhost # Hostname of current node
221
-Dmultinode.port=0 # Port for current node (0 = automatic)
222
-Dmultinode.server-host=localhost # Hostname of conductor node
223
-Dmultinode.server-port=4711 # Port of conductor node
224
-Dmultinode.index=0 # Index of current node (0-based)
225
```
226
227
### Required Akka Configuration
228
229
```hocon
230
akka {
231
actor.provider = remote
232
remote.artery.canonical {
233
hostname = ${multinode.host}
234
port = ${multinode.port}
235
}
236
testconductor {
237
barrier-timeout = 30s
238
query-timeout = 5s
239
}
240
}
241
```
242
243
### Network Failure Injection Setup
244
245
To use blackhole, passThrough, and throttle features:
246
247
```scala
248
object MyMultiNodeConfig extends MultiNodeConfig {
249
// Enable test transport for failure injection
250
testTransport(on = true)
251
}
252
```
253
254
## Integration Patterns
255
256
### ScalaTest Integration
257
258
```scala
259
trait STMultiNodeSpec extends MultiNodeSpecCallbacks
260
with AnyWordSpecLike with Matchers with BeforeAndAfterAll {
261
262
override def beforeAll() = multiNodeSpecBeforeAll()
263
override def afterAll() = multiNodeSpecAfterAll()
264
}
265
266
class MyTest extends MultiNodeSpec(MyConfig) with STMultiNodeSpec {
267
// Test implementation
268
}
269
```
270
271
### Performance Profiling
272
273
```scala
274
class MyTest extends MultiNodeSpec(MyConfig) with PerfFlamesSupport {
275
"performance test" in {
276
runPerfFlames(first, second)(5.seconds)
277
// Test code that will be profiled after 5 second delay
278
}
279
}
280
```