akka-core Documentation

Technical Documentation: akka/akka-core

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (`README.md`) Copy Section

Akka

Akka is a powerful platform that simplifies building and operating highly responsive, resilient, and scalable services.

The platform consists of

• the Akka SDK for straightforward, rapid development with AI assist and automatic clustering. Services built with the Akka SDK are automatically clustered and can be deployed on any infrastructure.
• and Akka Automated Operations, a managed solution that handles everything for Akka SDK services from auto-elasticity to multi-region high availability running safely within your VPC.

The Akka SDK and Akka Automated Operations are built upon the foundational Akka libraries, providing the building blocks for distributed systems.

Akka has been downloaded more than 1 billion times and has powered thousands of systems for over 15 years. Akka enables millions of concurrent users, terabyte stream processing, low-latency read/write data access, 99.9999% availability, and multi-region high availability and disaster recovery (HA/DR).

Akka is used to build agentic AI, AI inference, transactional, analytical, digital twin, IOT, and edge-to-cloud systems. It is simple to learn and operate. Akka creates production-ready systems without requiring prior knowledge of events, threading, or distributed systems.

Akka core library

The Akka core library provides:

• Multi-threaded behavior without the use of low-level concurrency constructs like
  atomics or locks — relieving you from even thinking about memory visibility issues.
• Transparent remote communication between systems and their components — relieving you from writing and maintaining difficult networking code.
• A clustered, high-availability architecture that is elastic, scales in or out, on demand — enabling you to deliver a truly reactive system.

At Akka's core is the actor model which provides a level of abstraction that makes it
easier to write correct concurrent, parallel and distributed systems. The actor
model spans the full set of Akka libraries, providing you with a consistent way
of understanding and using them. Thus, Akka offers a depth of integration that
you cannot achieve by picking libraries to solve individual problems and trying
to piece them together.

Using the Actor Model we raise the abstraction level and provide a better platform to build correct concurrent and scalable applications. This model is a perfect match for the principles laid out in the Reactive Manifesto.

For resilience, we adopt the "Let it crash" model which the telecom industry has used with great success to build applications that self-heal and systems that never stop.

Akka actors also provide the abstraction for transparent distribution and the basis for truly scalable and fault-tolerant applications.

Reference Documentation Copy Chapter

The current versions of all Akka libraries are listed on the Akka Dependencies page. Releases of the Akka core libraries in this repository are listed on the GitHub releases page.

The reference documentation for all Akka libraries is available via doc.akka.io/libraries/, details for the Akka core libraries
for Scala and Java.

The current versions of all Akka libraries are listed on the Akka Dependencies page. Releases of the Akka core libraries in this repository are listed on the GitHub releases page.

Contributing Copy Chapter

Contributions are very welcome!

If you see an issue that you'd like to see fixed, or want to shape out some ideas,
the best way to make it happen is to help out by submitting a pull request implementing it.
We welcome contributions from all, even you are not yet familiar with this project,
We are happy to get you started, and will guide you through the process once you've submitted your PR.

Refer to the CONTRIBUTING.md file for more details about the workflow,
and general hints on how to prepare your pull request. You can also ask for clarifications or guidance in GitHub issues directly,
or in the akka/dev chat if a more real time communication would be of benefit.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Tests and documentation are under a separate license, see the LICENSE file in each documentation and test root directory for details.

Overview (`akka-bench-jmh/README.md`) Copy Section

Akka Microbenchmarks

This subproject contains some microbenchmarks excercising key parts of Akka. (Excluding typed which has its
own jmh module)

You can run them like:

sbt shell
akka > project akka-bench-jmh
sbt:akka-bench-jmh> jmh:run -i 3 -wi 3 -f 1 .*ActorCreationBenchmark

or execute in one-line command

sbt akka-bench-jmh/jmh:run -i 3 -wi 3 -f 1 .*ActorCreationBenchmark

Use 'jmh:run -h' to get an overview of the available options.

Some potentially out of date resources for writing JMH benchmarks:

• Studying what's wrong with JMH benchmarks
• Writing good benchmarks

Overview (`artifact-bom/README.md`) Copy Section

artifact-bom

The Maven dependency descriptors in the artifact-bom directory
support our security tooling. These are generated by explicit activation of
the sbt-artifact-bom plugin on the
sbt projects.

File generation runs on compile. It should be enough to add those generated files
to the PR when changing the dependencies.

The generated file is then scanned by Wiz. Wiz status page for this project

Overview (`native-image-tests/README.md`) Copy Section

Projects used to verify native-image support works, run by CI or locally.

• scala-local local actor system app in Scala, exercising as many single-system features as possible
• scala-cluster clustered actor system app in Scala, exercising cluster and cluster tools

Running locally Copy Chapter

sbt publishLocal akka itself.

Build test project with sbt -Dakka.version=[local-snapshot-version] nativeImage and then start the generated native
image.

Overview (`samples/akka-quickstart-java/README.md`) Copy Section

Akka Quickstart with Java

Quickstart guide Copy Chapter

This example is described in https://doc.akka.io/libraries/akka-core/current/typed/actors.html#first-example

The Hello World example illustrates Akka basics. Within 10 minutes, you should be able to download and run the example and read the Introduction to Actors documentation to understand how the example is constructed. This will get your feet wet, and hopefully inspire you to dive deeper into the wonderful sea of Akka!

How to run Copy Chapter

This template also provides build descriptors for maven and gradle. You can use any of the following commands to run the application:

mvn compile exec:exec

or

./gradlew run

Template license Copy Chapter

Written in 2017 by Lightbend, Inc.

To the extent possible under law, the author(s) have dedicated all copyright and related
and neighboring rights to this template to the public domain worldwide.
This template is distributed without any warranty. See http://creativecommons.org/publicdomain/zero/1.0/.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-cluster-java/README.md`) Copy Section

This tutorial contains 3 samples illustrating different Akka cluster features.

• Subscribe to cluster membership events
• Sending messages to actors running on nodes in the cluster
• Cluster aware routers

A Simple Cluster Example Copy Chapter

To try this example locally, download the sources files with akka-samples-cluster-java.zip.

Open application.conf

To enable cluster capabilities in your Akka project you should, at a minimum, add the remote settings, and use cluster as the akka.actor.provider. The akka.cluster.seed-nodes should normally also be added to your application.conf file.

The seed nodes are configured contact points which newly started nodes will try to connect with in order to join the cluster.

Note that if you are going to start the nodes on different machines you need to specify the ip-addresses or host names of the machines in application.conf instead of 127.0.0.1.

Open SimpleClusterApp.java.

The small program together with its configuration starts an ActorSystem with the Cluster enabled. It joins the cluster and starts an actor that logs some membership events. Take a look at the SimpleClusterListener.java actor.

You can read more about the cluster concepts in the documentation.

To run this sample, first make sure everything is compiled using mvn compile, then run using mvn exec:java -Dexec.mainClass="sample.cluster.simple.App".

sample.cluster.simple.App starts three actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and then open three terminal windows.

In the first terminal window, start the first seed node with the following command:

mvn exec:java -Dexec.mainClass="sample.cluster.simple.App" -Dexec.args=25251

25251 corresponds to the port of the first seed-nodes element in the configuration. In the log output you see that the cluster node has been started and changed status to 'Up'.

In the second terminal window, start the second seed node with the following command:

mvn exec:java -Dexec.mainClass="sample.cluster.simple.App" -Dexec.args=25252

25252 corresponds to the port of the second seed-nodes element in the configuration. In the log output you see that the cluster node has been started and joins the other seed node and becomes a member of the cluster. Its status changed to 'Up'.

Switch over to the first terminal window and see in the log output that the member joined.

Start another node in the third terminal window with the following command:

mvn exec:java -Dexec.mainClass="sample.cluster.simple.App" -Dexec.args=0

Now you don't need to specify the port number, 0 means that it will use a random available port. It joins one of the configured seed nodes. Look at the log output in the different terminal windows.

Start even more nodes in the same way, if you like.

Shut down one of the nodes by pressing 'ctrl-c' in one of the terminal windows. It will cause the node to do a graceful leave from the cluster, telling the other nodes in the cluster that it is leaving. It will then be removed from the cluster, which you can see in the log output in the other terminals.

Look at the source code of the actor again. It registers itself as subscriber of certain cluster events. It gets notified a stream of events leading up to the current state. After that it receives events for changes that happen in the cluster.

Now we have seen how to subscribe to cluster membership events. You can read more about it in the documentation. The membership events show us the state of the cluster but it does not help with accessing actors on other nodes the cluster. To do that we need to use the Receptionist.

Worker registration example Copy Chapter

The Receptionist is a service registry that will work both when in single JVM apps not using cluster, and in clustered apps.
ActorRefs are registered to the receptionist using a ServiceKey. The service key is defined with a type of message that actors registered for it will accept and a string identifier.

Let's take a look at an example that illustrates how workers, here only on nodes with the role backend, register themselves to the receptionist so that frontend nodes will know what workers are available to perform their work. Note that a node could potentially have both roles, since the node roles are a set. The main provided only allows one role though.

The example application provides a service to transform text. At a periodic interval the frontend simulates an external request to process a text which it forwards to available workers if there are any.

Since the discovery of workers is dynamic both backend and frontend nodes can be added to the cluster dynamically.

The backend worker that performs the transformation job is defined in Worker.java. When starting up a worker registers itself to the receptionist so that it can be discovered through its ServiceKey on any node in the cluster.

The frontend that simulates user jobs as well as keeping track of available workers is defined in Frontend.java. The actor subscribes to the Receptionist with the WorkerServiceKey to receive updates when the set of available workers in the cluster changes. If a worker dies or its node is removed from the cluster the receptionist will send out an updated listing so the frontend does not need to watch the workers.

To run this sample, make sure you have shut down any previously started cluster sample, then type mvn exec:java -Dexec.mainClass="sample.cluster.transformation.App".

TransformationApp starts 5 actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and run the following commands in separate terminal windows.

mvn exec:java -Dexec.mainClass="sample.cluster.transformation.App" -Dexec.args="backend 25251"

mvn exec:java -Dexec.mainClass="sample.cluster.transformation.App" -Dexec.args="backend 25252"

mvn exec:java -Dexec.mainClass="sample.cluster.transformation.App" -Dexec.args="backend 0"

mvn exec:java -Dexec.mainClass="sample.cluster.transformation.App" -Dexec.args="frontend 0"

mvn exec:java -Dexec.mainClass="sample.cluster.transformation.App" -Dexec.args="frontend 0"

There is a component built into Akka that performs the task of subscribing to the receptionist and keeping track of available actors significantly simplifying such interactions: the group router. Let's look into how we can use those in the next section!

Cluster Aware Routers Copy Chapter

The group routers relies on the Receptionist and will therefore route messages to services registered in any node of the cluster.

Let's take a look at a few samples that make use of cluster aware routers.

Cluster routing example Copy Chapter

Let's take a look at two different ways to distribute work across a cluster using routers.

Note that the samples just shows off various parts of Akka Cluster and does not provide a complete structure to build a resilient distributed application with.

Example with Group of routees Copy Section

The example application provides a service to calculate statistics for a text. When some text is sent to the service it splits it into words, and delegates the task to count number of characters in each word to a separate worker, a routee of a router. The character count for each word is sent back to an aggregator that calculates the average number of characters per word when all results have been collected.

The worker that counts number of characters in each word is defined in StatsWorker.java.

The service that receives text from users and splits it up into words, delegates to a pool of workers and aggregates the result is defined in StatsService.java.

Note, nothing cluster specific so far, just plain actors.

Nodes in the cluster can be marked with roles, to perform different tasks, in our case we use compute as a role to
designate cluster nodes that should do processing of word statistics.

In StatsSample.java each compute node starts a StatsService
that distributes work over N local StatsWorkers. The client nodes then message the StatsService instances through a group router.
The router finds services by subscribing to the cluster receptionist and a service key. Each worker is registered to the receptionist
when started.

With this design a single compute node crashing will only lose the ongoing work in that node and have the other nodes
keep on with their work, but there is no single place to ask for a list of the current work in progress.

To run the sample, type mvn exec:java -Dexec.mainClass="sample.cluster.stats.App" if it is not already started.

StatsSample starts 4 actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and run the following commands in separate terminal windows.

mvn exec:java -Dexec.mainClass="sample.cluster.stats.App" -Dexec.args="compute 25251"

mvn exec:java -Dexec.mainClass="sample.cluster.stats.App" -Dexec.args="compute 25252"

mvn exec:java -Dexec.mainClass="sample.cluster.stats.App" -Dexec.args="compute 0"

mvn exec:java -Dexec.mainClass="sample.cluster.stats.App" -Dexec.args="client 0"

Router example with Cluster Singleton Copy Section

AppOneMaster.java each compute node starts
N workers, that register themselves with the receptionist. The StatsService is run in a single instance in the cluster
through the Akka Cluster Singleton. The actual work is performed by workers on all compute nodes though. The workers
are reached through a group router used by the singleton.

With this design it would be possible to query the singleton for current work - it knows all current requests in flight
and could potentially make decisions based on knowing exactly what work is currently in progress.

If the singleton node crashes however, all ongoing work is lost though since the state of the singleton is not persistent, when it is started on a new node the StatsService will not know of any previous work. It also means that since all work has to go through the singleton it could be come a bottleneck. If one of the other nodes crash only the ongoing work sent to them is lost, however since each ongoing request could be handled by multiple different workers on different nodes a crash could cause problems to many requests.

To run this sample, type mvn exec:java -Dexec.mainClass="sample.cluster.stats.AppOneMaster" if it is not already started.

AppOneMaster starts 4 actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and run the following commands in separate terminal windows.

mvn exec:java -Dexec.mainClass="sample.cluster.stats.AppOneMaster" -Dexec.args="compute 25251"

mvn exec:java -Dexec.mainClass="sample.cluster.stats.AppOneMaster" -Dexec.args="compute 25252"

mvn exec:java -Dexec.mainClass="sample.cluster.stats.AppOneMaster" -Dexec.args="compute 0"

mvn exec:java -Dexec.mainClass="sample.cluster.stats.AppOneMaster" -Dexec.args="client 0"

Tests Copy Chapter

The multi-jvm testkit which allows for starting a cluster with multiple separate JVMs is only available from Scala with
the Scala build tool sbt. Such tests are included for completeness and can be found in src/multi-jvm.
You can run them by typing sbt multi-jvm:test.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-cluster-scala/README.md`) Copy Section

This tutorial contains 3 samples illustrating different Akka cluster features.

• Subscribe to cluster membership events
• Sending messages to actors running on nodes in the cluster
• Cluster aware routers

A Simple Cluster Example Copy Chapter

To try this example locally, download the sources files with akka-samples-cluster-scala.zip.

Open application.conf

To enable cluster capabilities in your Akka project you should, at a minimum, add the remote settings, and use cluster as the akka.actor.provider. The akka.cluster.seed-nodes should normally also be added to your application.conf file.

The seed nodes are configured contact points which newly started nodes will try to connect with in order to join the cluster.

Note that if you are going to start the nodes on different machines you need to specify the ip-addresses or host names of the machines in application.conf instead of 127.0.0.1.

Open SimpleClusterApp.scala.

The small program together with its configuration starts an ActorSystem with the Cluster enabled. It joins the cluster and starts an actor that logs some membership events. Take a look at the SimpleClusterListener.scala actor.

You can read more about the cluster concepts in the documentation.

To run this sample, type sbt "runMain sample.cluster.simple.App".

sample.cluster.simple.App starts three actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and then open three terminal windows.

In the first terminal window, start the first seed node with the following command:

sbt "runMain sample.cluster.simple.App 25251"

25251 corresponds to the port of the first seed-nodes element in the configuration. In the log output you see that the cluster node has been started and changed status to 'Up'.

In the second terminal window, start the second seed node with the following command:

sbt "runMain sample.cluster.simple.App 25252"

25252 corresponds to the port of the second seed-nodes element in the configuration. In the log output you see that the cluster node has been started and joins the other seed node and becomes a member of the cluster. Its status changed to 'Up'.

Switch over to the first terminal window and see in the log output that the member joined.

Start another node in the third terminal window with the following command:

sbt "runMain sample.cluster.simple.App 0"

Now you don't need to specify the port number, 0 means that it will use a random available port. It joins one of the configured seed nodes. Look at the log output in the different terminal windows.

Start even more nodes in the same way, if you like.

Shut down one of the nodes by pressing 'ctrl-c' in one of the terminal windows. It will cause the node to do a graceful leave from the cluster, telling the other nodes in the cluster that it is leaving. It will then be removed from the cluster, which you can see in the log output in the other terminals.

Look at the source code of the actor again. It registers itself as subscriber of certain cluster events. It gets notified a stream of events leading up to the current state. After that it receives events for changes that happen in the cluster.

Now we have seen how to subscribe to cluster membership events. You can read more about it in the documentation. The membership events show us the state of the cluster but it does not help with accessing actors on other nodes the cluster. To do that we need to use the Receptionist.

Worker registration example Copy Chapter

The Receptionist is a service registry that will work both when in single JVM apps not using cluster, and in clustered apps.
ActorRefs are registered to the receptionist using a ServiceKey. The service key is defined with a type of message that actors registered for it will accept and a string identifier.

Let's take a look at an example that illustrates how workers, here only on nodes with the role backend, register themselves to the receptionist so that frontend nodes will know what workers are available to perform their work. Note that a node could potentially have both roles, since the node roles are a set. The main provided only allows one role though.

The example application provides a service to transform text. At a periodic interval the frontend simulates an external request to process a text which it forwards to available workers if there are any.

Since the discovery of workers is dynamic both backend and frontend nodes can be added to the cluster dynamically.

The backend worker that performs the transformation job is defined in TransformationBackend.scala. When starting up a worker registers itself to the receptionist so that it can be discovered through its ServiceKey on any node in the cluster.

The frontend that simulates user jobs as well as keeping track of available workers is defined in Frontend.scala. The actor subscribes to the Receptionist with the WorkerServiceKey to receive updates when the set of available workers in the cluster changes. If a worker dies or its node is removed from the cluster the receptionist will send out an updated listing so the frontend does not need to watch the workers.

To run this sample, make sure you have shut down any previously started cluster sample, then type sbt "runMain sample.cluster.transformation.App".

TransformationApp starts 5 actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and run the following commands in separate terminal windows.

sbt "runMain sample.cluster.transformation.App backend 25251"

sbt "runMain sample.cluster.transformation.App backend 25252"

sbt "runMain sample.cluster.transformation.App backend 0"

sbt "runMain sample.cluster.transformation.App frontend 0"

sbt "runMain sample.cluster.transformation.App frontend 0"

There is a component built into Akka that performs the task of subscribing to the receptionist and keeping track of available actors significantly simplifying such interactions: the group router. Let's look into how we can use those in the next section!

Cluster Aware Routers Copy Chapter

The group routers relies on the Receptionist and will therefore route messages to services registered in any node of the cluster.

Let's take a look at a few samples that make use of cluster aware routers.

Cluster routing example Copy Chapter

Let's take a look at two different ways to distribute work across a cluster using routers.

Note that the samples just shows off various parts of Akka Cluster and does not provide a complete structure to build a resilient distributed application with.

Example with Group of routees Copy Section

The example application provides a service to calculate statistics for a text. When some text is sent to the service it splits it into words, and delegates the task to count number of characters in each word to a separate worker, a routee of a router. The character count for each word is sent back to an aggregator that calculates the average number of characters per word when all results have been collected.

The worker that counts number of characters in each word is defined in StatsWorker.scala.

The service that receives text from users and splits it up into words, delegates to a pool of workers and aggregates the result is defined in StatsService.scala.

Note, nothing cluster specific so far, just plain actors.

Nodes in the cluster can be marked with roles, to perform different tasks, in our case we use compute as a role to
designate cluster nodes that should do processing of word statistics.

In StatsSample.scala each compute node starts a StatsService
that distributes work over N local StatsWorkers. The client nodes then message the StatsService instances through a group router.
The router finds services by subscribing to the cluster receptionist and a service key. Each worker is registered to the receptionist
when started.

With this design a single compute node crashing will only lose the ongoing work in that node and have the other nodes
keep on with their work, but there is no single place to ask for a list of the current work in progress.

To run the sample, type sbt "runMain sample.cluster.stats.App" if it is not already started.

StatsSample starts 4 actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and run the following commands in separate terminal windows.

sbt "runMain sample.cluster.stats.App compute 25251"

sbt "runMain sample.cluster.stats.App compute 25252"

sbt "runMain sample.cluster.stats.App compute  0"

sbt "runMain sample.cluster.stats.App client 0"

Router example with Cluster Singleton Copy Section

StatsSampleOneMaster.scala each compute node starts
N workers, that register themselves with the receptionist. The StatsService is run in a single instance in the cluster
through the Akka Cluster Singleton. The actual work is performed by workers on all compute nodes though. The workers
are reached through a group router used by the singleton.

With this design it would be possible to query the singleton for current work - it knows all current requests in flight
and could potentially make decisions based on knowing exactly what work is currently in progress.

If the singleton node crashes however, all ongoing work is lost though since the state of the singleton is not persistent, when it is started on a new node the StatsService will not know of any previous work. It also means that since all work has to go through the singleton it could be come a bottleneck. If one of the other nodes crash only the ongoing work sent to them is lost, however since each ongoing request could be handled by multiple different workers on different nodes a crash could cause problems to many requests.

To run this sample, type sbt "runMain sample.cluster.stats.AppOneMaster" if it is not already started.

StatsSampleOneMaster starts 4 actor systems (cluster members) in the same JVM process. It can be more interesting to run them in separate processes. Stop the application and run the following commands in separate terminal windows.

sbt "runMain sample.cluster.stats.AppOneMaster compute 25251"

sbt "runMain sample.cluster.stats.AppOneMaster compute 25252"

sbt "runMain sample.cluster.stats.AppOneMaster compute 0"

sbt "runMain sample.cluster.stats.AppOneMaster client 0"

Tests Copy Chapter

Tests can be found in src/multi-jvm. You can run them by typing sbt multi-jvm:test.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-distributed-data-java/README.md`) Copy Section

This tutorial contains 5 samples illustrating how to use Akka Distributed Data.

• Low Latency Voting Service
• Highly Available Shopping Cart
• Distributed Service Registry
• Replicated Cache
• Replicated Metrics

To try this example locally, download the sources files with akka-sample-distributed-data-java.zip.

Akka Distributed Data is useful when you need to share data between nodes in an Akka Cluster. The data is accessed with an actor providing a key-value store like API. The keys are unique identifiers with type information of the data values. The values are Conflict Free Replicated Data Types (CRDTs).

All data entries are spread to all nodes, or nodes with a certain role, in the cluster via direct replication and gossip based dissemination. You have fine grained control of the consistency level for reads and writes.

The nature CRDTs makes it possible to perform updates from any node without coordination. Concurrent updates from different nodes will automatically be resolved by the monotonic merge function, which all data types must provide. The state changes always converge. Several useful data types for counters, sets, maps and registers are provided and you can also implement your own custom data types.

It is eventually consistent and geared toward providing high read and write availability (partition tolerance), with low latency. Note that in an eventually consistent system a read may return an out-of-date value.

Note that there are some Limitations that you should be aware of. For example, Akka Distributed Data is not intended for Big Data.

Low Latency Voting Service Copy Chapter

Distributed Data is great for low latency services, since you can update or get data from the local replica without immediate communication with other nodes.

Open VotingService.java.

VotingService is an actor for low latency counting of votes on several cluster nodes and aggregation of the grand total number of votes. The actor is started on each cluster node. First it expects an OPEN message on one or several nodes. After that the counting can begin. The open signal is immediately replicated to all nodes with a boolean Flag. Note writeAll.

Update<Flag> update = new Update<>(openedKey, Flag.create(), writeAll, curr -> curr.switchOn());

The actor is subscribing to changes of the OpenedKey and other instances of this actor, also on other nodes, will be notified when the flag is changed.

replicator.tell(new Subscribe<>(openedKey, self()), ActorRef.noSender());

.match(Changed.class, c -> c.key().equals(openedKey), c -> receiveOpenedChanged((Changed<Flag>) c))

The counters are kept in a PNCounterMap and updated with:

Update update = new Update<>(countersKey, PNCounterMap.create(), Replicator.writeLocal(),
        curr -> curr.increment(node, vote.participant, 1));
 replicator.tell(update, self());

Incrementing the counter is very fast, since it only involves communication with the local Replicator actor. Note writeLocal. Those updates are also spread to other nodes, but that is performed in the background.

The total number of votes is retrieved with:

Optional<Object> ctx = Optional.of(sender());
replicator.tell(new Replicator.Get(countersKey, readAll, ctx), self());

.match(GetSuccess.class, g -> g.key().equals(countersKey),
   g -> receiveGetSuccess(open, (GetSuccess) g))

private void receiveGetSuccess(boolean open, GetSuccess g) {
  Map<String, BigInteger> result = g.dataValue().getEntries();
  ActorRef replyTo = (ActorRef) g.getRequest().get();
  replyTo.tell(new Votes(result, open), self());
}

The multi-node test for the VotingService can be found in VotingServiceSpec.scala.

Read the Using the Replicator documentation for more details of how to use Get, Update, and Subscribe.

Highly Available Shopping Cart Copy Chapter

Distributed Data is great for highly available services, since it is possible to perform updates to the local node (or currently available nodes) during a network partition.

Open ShoppingCart.java.

ShoppingCart is an actor that holds the selected items to buy for a user. The actor instance for a specific user may be started where ever needed in the cluster, i.e. several instances may be started on different nodes and used at the same time.

Each product in the cart is represented by a LineItem and all items in the cart is collected in a LWWMap.

The actor handles the commands GET_CART, AddItem and RemoveItem. To get the latest updates in case the same shopping cart is used from several nodes it is using consistency level of readMajority and writeMajority, but that is only done to reduce the risk of seeing old data. If such reads and writes cannot be completed due to a network partition it falls back to reading/writing from the local replica (see GetFailure). Local reads and writes will always be successful and when the network partition heals the updated shopping carts will be be disseminated by the gossip protocol and the LWWMap CRDTs are merged, i.e. it is a highly available shopping cart.

The multi-node test for the ShoppingCart can be found in ShoppingCartSpec.scala.

Read the Consistency section in the documentation to understand the consistency considerations.

Replicated Cache Copy Chapter

This example illustrates a simple key-value cache.

Open ReplicatedCache.scala.

ReplicatedCache is an actor that is started on each node in the cluster. It supports three commands: PutInCache, GetFromCache and Evict.

It is splitting up the key space in 100 top level keys, each with a LWWMap. When a data entry is changed the full state of that entry is replicated to other nodes, i.e. when you update a map the whole map is replicated. Therefore, instead of using one ORMap with 1000 elements it is more efficient to split that up in 100 top level ORMap entries with 10 elements each. Top level entries are replicated individually, which has the trade-off that different entries may not be replicated at the same time and you may see inconsistencies between related entries. Separate top level entries cannot be updated atomically together.

The multi-node test for the ReplicatedCache can be found in ReplicatedCacheSpec.scala.

Replicated Metrics Copy Chapter

This example illustrates to spread metrics data to all nodes in an Akka cluster.

Open ReplicatedMetrics.java.

ReplicatedMetrics is an actor that is started on each node in the cluster. Periodically it collects some metrics, in this case used and max heap size. Each metrics type is stored in a LWWMap where the key in the map is the address of the node. The values are disseminated to other nodes with the gossip protocol.

The multi-node test for the ReplicatedCache can be found in ReplicatedMetricsSpec.scala.

Note that there are some Limitations that you should be aware of. For example, Akka Distributed Data is not intended for Big Data.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-distributed-data-scala/README.md`) Copy Section

This tutorial contains 5 samples illustrating how to use Akka Distributed Data.

• Low Latency Voting Service
• Highly Available Shopping Cart
• Distributed Service Registry
• Replicated Cache
• Replicated Metrics

To try this example locally, download the sources files with akka-sample-distributed-data-scala.zip.

Akka Distributed Data is useful when you need to share data between nodes in an Akka Cluster. The data is accessed with an actor providing a key-value store like API. The keys are unique identifiers with type information of the data values. The values are Conflict Free Replicated Data Types (CRDTs).

All data entries are spread to all nodes, or nodes with a certain role, in the cluster via direct replication and gossip based dissemination. You have fine grained control of the consistency level for reads and writes.

The nature CRDTs makes it possible to perform updates from any node without coordination. Concurrent updates from different nodes will automatically be resolved by the monotonic merge function, which all data types must provide. The state changes always converge. Several useful data types for counters, sets, maps and registers are provided and you can also implement your own custom data types.

It is eventually consistent and geared toward providing high read and write availability (partition tolerance), with low latency. Note that in an eventually consistent system a read may return an out-of-date value.

Note that there are some Limitations that you should be aware of. For example, Akka Distributed Data is not intended for Big Data.

Low Latency Voting Service Copy Chapter

Distributed Data is great for low latency services, since you can update or get data from the local replica without immediate communication with other nodes.

Open VotingService.scala.

VotingService is an actor for low latency counting of votes on several cluster nodes and aggregation of the grand total number of votes. The actor is started on each cluster node. First it expects an Open message on one or several nodes. After that the counting can begin. The open signal is immediately replicated to all nodes with a boolean Flag. Note WriteAll.

replicator ! Update(OpenedKey, Flag(), WriteAll(5.seconds))(_.switchOn)

The actor is subscribing to changes of the OpenedKey and other instances of this actor, also on other nodes, will be notified when the flag is changed.

replicator ! Subscribe(OpenedKey, self)

case c @ Changed(OpenedKey) if c.get(OpenedKey).enabled

The counters are kept in a PNCounterMap and updated with:

val update = Update(CountersKey, PNCounterMap(), WriteLocal, request = Some(v)) {
  _.increment(participant, 1)
}
replicator ! update

Incrementing the counter is very fast, since it only involves communication with the local Replicator actor. Note WriteLocal. Those updates are also spread to other nodes, but that is performed in the background.

The total number of votes is retrieved with:

case GetVotes ⇒
  replicator ! Get(CountersKey, ReadAll(3.seconds), Some(GetVotesReq(sender())))

case g @ GetSuccess(CountersKey, Some(GetVotesReq(replyTo))) ⇒
  val data = g.get(CountersKey)
  replyTo ! Votes(data.entries, open)

The multi-node test for the VotingService can be found in VotingServiceSpec.scala.

Read the Using the Replicator documentation for more details of how to use Get, Update, and Subscribe.

Highly Available Shopping Cart Copy Chapter

Distributed Data is great for highly available services, since it is possible to perform updates to the local node (or currently available nodes) during a network partition.

Open ShoppingCart.scala.

ShoppingCart is an actor that holds the selected items to buy for a user. The actor instance for a specific user may be started where ever needed in the cluster, i.e. several instances may be started on different nodes and used at the same time.

Each product in the cart is represented by a LineItem and all items in the cart is collected in a LWWMap.

The actor handles the commands GetCart, AddItem and RemoveItem. To get the latest updates in case the same shopping cart is used from several nodes it is using consistency level of ReadMajority and WriteMajority, but that is only done to reduce the risk of seeing old data. If such reads and writes cannot be completed due to a network partition it falls back to reading/writing from the local replica (see GetFailure). Local reads and writes will always be successful and when the network partition heals the updated shopping carts will be be disseminated by the gossip protocol and the LWWMap CRDTs are merged, i.e. it is a highly available shopping cart.

The multi-node test for the ShoppingCart can be found in ShoppingCartSpec.scala.

Read the Consistency section in the documentation to understand the consistency considerations.

Replicated Cache Copy Chapter

This example illustrates a simple key-value cache.

Open ReplicatedCache.scala.

ReplicatedCache is an actor that is started on each node in the cluster. It supports three commands: PutInCache, GetFromCache and Evict.

It is splitting up the key space in 100 top level keys, each with a LWWMap. When a data entry is changed the full state of that entry is replicated to other nodes, i.e. when you update a map the whole map is replicated. Therefore, instead of using one ORMap with 1000 elements it is more efficient to split that up in 100 top level ORMap entries with 10 elements each. Top level entries are replicated individually, which has the trade-off that different entries may not be replicated at the same time and you may see inconsistencies between related entries. Separate top level entries cannot be updated atomically together.

The multi-node test for the ReplicatedCache can be found in ReplicatedCacheSpec.scala.

Replicated Metrics Copy Chapter

This example illustrates to spread metrics data to all nodes in an Akka cluster.

Open ReplicatedMetrics.scala.

ReplicatedMetrics is an actor that is started on each node in the cluster. Periodically it collects some metrics, in this case used and max heap size. Each metrics type is stored in a LWWMap where the key in the map is the address of the node. The values are disseminated to other nodes with the gossip protocol.

The multi-node test for the ReplicatedCache can be found in ReplicatedMetricsSpec.scala.

Note that there are some Limitations that you should be aware of. For example, Akka Distributed Data is not intended for Big Data.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-fsm-java/README.md`) Copy Section

Finite State Machine in Actors Copy Chapter

This sample is an adaptation of Dining Hakkers.

To try this example locally, download the sources files with akka-samples-fsm-java.zip.

Open DiningHakkersTyped.scala.

It illustrates how the behaviors and transitions can be defined with Akka Typed.

Start the application by typing mvn compile exec:java -Dexec.mainClass="sample.DiningHakkers". In the log output you can see the actions of the Hakker actors.

Read more about Akka Typed in the documentation.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-fsm-scala/README.md`) Copy Section

Finite State Machine in Actors Copy Chapter

This sample is an adaptation of Dining Hakkers.

To try this example locally, download the sources files with akka-samples-fsm-scala.zip.

Open DiningHakkersTyped.scala.

It illustrates how the behaviors and transitions can be defined with Akka Typed.

Start the application by typing sbt "runMain sample.DiningHakkers". In the log output you can see the actions of the Hakker actors.

Read more about Akka Typed in the documentation.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-kafka-to-sharding-scala/README.md`) Copy Section

Aligning Kafka Partitions with Akka Cluster Sharding

It is common to consume a Kafka topic and forward the messages to sharded actors.

The Kafka consumer can be started on each node with the same group id
and then the messages forwarded to sharding via an ask. It is important to use ask
rather than tell to enable backpressure from the sharded actor to the Kafka consumer.

Using the default shard allocation strategy there is no relation between the Kafka partitions
allocated to a consumer and the location of the shards meaning that most messages will
have one network hop.

If all of the messages for the same sharded entity are in the same Kafka partition then
this can be improved on with the external shard allocation strategy.
For this to be true the producer partitioning must align with the shard extraction
in cluster sharding.

Imagine a scenario that processes all events for users with following constraints:

• The key of the kafka message is the user id which is in turn the entity id in sharding
• All messages for the same user id end up in the same partition

Then we can enforce that the kafka partition == the akka cluster shard id and use the external
sharding allocation strategy to move shards to the node that is consuming that partition, resulting
in no cross node traffic.

Read the following documentation to learn more about Akka Cluster External Shard Allocation
and its support for Kafka in Alpakka Kafka Cluster Sharding.

Running the sample

The sample is made up of three applications:

• producer A Kafka producer, that produces events about users
• processor An Akka Cluster Sharding application that reads the Kafka topic and forwards the messages to a sharded
  entity that represents a user and a gRPC front end for accessing the sharded actors state
• client A gRPC client for interacting with the cluster
• kafka A local Kafka server

The sample demonstrates how the external shard allocation strategy can be used so messages are processed locally.

The sample depends on a Kafka broker running locally on port 9092 with a topic with 128 partitions called user-events.
Kafka can be run in Docker or run locally using the optional kafka project.

• Run the local Kafka server. This project will also create the user-events topic.

sbt "kafka / run"

In the Kafka server window you'll see the following when the server is ready:

12:06:59.711 INFO  [run-main-0          ] s.s.embeddedkafka.KafkaBroker$        Kafka running: localhost:9092
12:06:59.711 INFO  [run-main-0          ] s.s.embeddedkafka.KafkaBroker$        Topic 'user-events' with 128 partitions created

If you want to use a different Kafka cluster then update the applications.confs in each project to point to your
Kafka broker if not running on localhost:9092.

• (Optional) If you do not run the local Kafka server then you must create a topic with 128 partitions, or update
  application.conf with the desired number of partitions e.g. a command from your Kafka installation:

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 128 --topic user-events

• Start a single processor, this will consume the messages from the topic and distribute them to sharding,
  three arguments are required, the akka remoting port, the akka management port, and the gRPC port for the front end.
• If you run on different ports the first two akka remoting ports should be 2551/2552 as they are configured as seeds.
• As there is a single consumer, all partitions will initially be assigned to this node.

sbt "processor / run 2551 8551 8081"

The processor starts a KafkaConsumer, as it is the only consumer in the group it will be assigned every single Kafka partition
and shards for each partition will be assigned to the current node. You will see logs like:

[info] [2020-01-16 09:48:51,040] [INFO] [akka://KafkaToSharding/user/kafka-event-processor/rebalancerRef] - Partition [1] assigned to current node. Updating shard allocation
[info] [2020-01-16 09:48:51,040] [INFO] [akka://KafkaToSharding/user/kafka-event-processor/rebalancerRef] - Partition [25] assigned to current node. Updating shard allocation
[info] [2020-01-16 09:48:51,043] [INFO] [akka://KafkaToSharding/user/kafka-event-processor/rebalancerRef] - Partition [116] assigned to current node. Updating shard allocation

If there are existing messages on the topic they will all be processed locally as there is a single node.

Next we start the Kafka producer to see some messages flowing from Kafka to sharding.

sbt "producer / run"

In the producer window you'll see:

[INFO] [01/16/2020 09:51:38.639] [UserEventProducer(akka://UserEventProducer)] Sending message to user 29
[INFO] [01/16/2020 09:51:39.660] [UserEventProducer(akka://UserEventProducer)] Sending message to user 60
[INFO] [01/16/2020 09:51:40.680] [UserEventProducer(akka://UserEventProducer)] Sending message to user 75

In the single processor node the messages will start flowing:

/* Detailed source-code truncated for AI context efficiency. */

The first log line is just after the message has been taken from Kafka.
The second log is from the sharded entity. The goal is to have these
always on the same node as the external shard allocation strategy will move the shard to where ever the
Kafka partition is being consumed.

As there is only one node we get 100% locality, each forwarded message is processed on the same node

Now let's see that remain true once we add more nodes to the Akka Cluster, add another with different ports:

sbt "processor / run 2552 8552 8082"

When this starts up we'll see Kafka assign partitions to the new node (it is in the same consumer group):

Partition [29] assigned to current node. Updating shard allocation

On one of the nodes, where the ShardCoordinator runs, we'll see the rebalance happening:

[info] [2020-01-16 09:59:39,923] [INFO] [akka://[email protected]:2551/system/sharding/user-processingCoordinator/singleton/coordinator] - Starting rebalance for shards [45,33,16,2,3,15,11,6,36]. Current shards rebalancing: []

Both nodes now have roughly 64 shards / partitions, all co-located with the Kafka Consuemer.
You can verify this by the logs showing that when a message is received by the Kafka Consumer when it is forwarded to
cluster sharding the entity logs receiving the event on the same node.

[info] [2020-01-17 08:27:58,199] [INFO] [akka://KafkaToSharding/user/kafka-event-processor] - Forwarding message for entity 29 to cluster sharding
[info] [2020-01-17 08:27:58,204] [INFO] [akka://KafkaToSharding/system/sharding/user-processing/45/29] - user 29 purchase cat t-shirt, quantity 1, price 2093
[info] [2020-01-17 08:28:08,218] [INFO] [akka://KafkaToSharding/user/kafka-event-processor] - Forwarding message for entity 56 to cluster sharding
[info] [2020-01-17 08:28:08,218] [INFO] [akka://KafkaToSharding/system/sharding/user-processing/6/56] - user 56 purchase akka t-shirt, quantity 3, price 8576
[info] [2020-01-17 08:28:28,288] [INFO] [akka://KafkaToSharding/user/kafka-event-processor] - Forwarding message for entity 44 to cluster sharding
[info] [2020-01-17 08:28:28,296] [INFO] [akka://KafkaToSharding/system/sharding/user-processing/59/44] - user 44 purchase cat t-shirt, quantity 3, price 9716

Each forwarding messaging is followed by log for the same entity on the current node.

Using Akka management we can see the shard allocations and the number of entities per shard (uses curl and jq):

# Node 1:
curl -v localhost:8551/cluster/shards/user-processing | jq

# Node 2:
curl -v localhost:8552/cluster/shards/user-processing | jq

We can count the number of shards on each:

# Node 1:
curl -s localhost:8551/cluster/shards/user-processing | jq -r "." | grep shardId | wc -l
# Node 2:
curl -s localhost:8552/cluster/shards/user-processing | jq -r "." | grep shardId | wc -l

The number of shards will depend on which entities have received messages.

We now have a 2 node Akka Cluster with a Kafka Consumer running on each where the kafka partitions align
with Cluster shards.

A use case for sending the processing to sharding is it allows each entity to be queried from any where in the cluster
e.g. from a HTTP or gRPC front end.

The sample includes a gRPC front end that gets the running total of number of purchases and total money spent
by each customer. Requests can come via gRPC on any node for any entity but sharding will route them to
the correct node even if that moves due to a kafka rebalance.

A gRPC client is included which can be started with...

sbt "client / run"

It assumes there is one of the nodes running its front end port on 8081.
You can enter the user id to get their stats: The users are 0-199, try entering users
that have shown up in the logs of the other processes.

7
User 7 has made 2 for a total of 3096p
Enter user id or :q to quit
3
User 3 has made 1 for a total of 12060p
Enter user id or :q to quit
4
User 4 has made 1 for a total of 7876p
Enter user id or :q to quit
5
User 5 has made 0 for a total of 0p
Enter user id or :q to quit
1
User 1 has made 0 for a total of 0p
Enter user id or :q to quit

We've now demonstrated two things:

• Keeping the processing local, where ever the Kafka partition is consumed the shard will be moved to that location
• The state for each entity is globally accessible from all nodes

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-sharding-java/README.md`) Copy Section

Cluster Sharding sample

The KillrWeather sample illustrates how to use Akka Cluster Sharding in Java.
It also shows the basic usage of Akka HTTP.

To try this example locally, download the sources files with akka-sample-sharding-java.zip.

The sample consists of two applications, each a separate maven submodule:

• killrweather - A distributed Akka cluster that shards weather stations, each keeping a set of recorded
  data points and allowing for local querying of the records. The cluster app has a HTTP endpoint for recording
  and querying the data per weather station.

• killrweather-fog - A client that periodically submits random weather measurements for a set of stations to a
  running cluster.

Let's start going through the implementation of the cluster!

KillrWeather Copy Chapter

Open KillrWeather.java.
This program starts an ActorSystem which joins the cluster through configuration, and starts a Guardian actor for the system.

Guardian Copy Section

The Guardian.java bootstraps the application to shard
WeatherStation actors across the cluster nodes.

Setting up sharding with the entity is done in WeatherStation.java.
Keeping the setup logic together with the sharded actor and then calling it from the bootstrap logic of the application
is a common pattern to structure sharded entities.

WeatherStation - sharded data by id Copy Section

Each sharded WeatherStation actor has a set of recorded data points for a station identifier.

It receives that data stream from remote devices via the
HTTP endpoint. Each WeatherStation and also respond to queries about it's recorded set of data such as:

• current
• averages
• high/low

Receiving edge device data by data type Copy Section

The WeatherHttpServer is started with
WeatherRoutes
to receive and unmarshall data from remote devices by station ID to allow
querying. To interact with the sharded entities it uses the EntityRef API.

The HTTP port of each node is chosen from the port used for Akka Remoting plus 10000, so for a node running
on port 2525 the HTTP port will be 12525.

Configuration Copy Section

This application is configured in killrweather/src/main/resources/application.conf
Before running, first make sure the correct settings are set for your system, as described in the akka-sample-cluster tutorial.

Fog Network Copy Chapter

Open Fog.java.

Fog is the program simulating many weather stations and their devices which read and report data to clusters.
The name refers to Fog computing with edges - the remote weather station
nodes and their device edges.

This example starts simply with one actor per station and just reports one data type, temperature. In the wild, other devices would include:
pressure, precipitation, wind speed, wind direction, sky condition and dewpoint.
Fog starts the number of weather stations configured in killrweather-fog/src/main/resources/application.conf
upon boot.

Weather stations and devices Copy Section

Each WeatherStation is run on a task to trigger scheduled data sampling.
These samples are timestamped and sent to the cluster over HTTP using Akka HTTP.

Akka HTTP example Copy Chapter

Within KillrWeather are two simple sides to an HTTP equation.

Client

• WeatherStation - HTTP data marshall and send

Server

• WeatherHttpServer - HTTP server
• WeatherRoutes - HTTP routes receiver which will unmarshall and pass on the data

Both parts of the application uses Jackson for marshalling and unmarshalling objects to and from JSON.

Running the samples Copy Chapter

The KillrWeather Cluster Copy Section

There are two ways to run the cluster, the first is a convenience quick start.

A simple three node cluster in the same JVM

The simplest way to run this sample is to run this in a terminal, if not already started:

mvn compile
mvn -pl killrweather exec:java

This command starts three (the default) KillrWeather actor systems (a three node cluster) in the same JVM process.

Dynamic WeatherServer ports

In the log snippet below, note the dynamic weather ports opened by each KillrWeather node's WeatherServer for weather stations to connect to.
The number of ports are by default three, for the minimum three node cluster. You can start more cluster nodes, so these are dynamic to avoid bind errors.

[2020-01-16 13:44:58,842] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-3] [] - WeatherServer online at http://127.0.0.1:12553/
[2020-01-16 13:44:58,842] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-19] [] - WeatherServer online at http://127.0.0.1:53937/
[2020-01-16 13:44:58,843] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-15] [] - WeatherServer online at http://127.0.0.1:12554/

A three node cluster in separate JVMs

It is more interesting to run them in separate processes. Stop the application and then open three terminal windows.
In the first terminal window, start the first seed node with the following command:

mvn -pl killrweather exec:java -Dexec.args="2553"

2553 corresponds to the port of the first seed-nodes element in the configuration. In the log output you see that the cluster node has been started and changed status to 'Up'.

You'll see a log message when a WeatherStation sends a message to record the current temperature, and for each of those you'll see a log message from the WeatherRoutes showing the action taken and the new average temperature.

In the second terminal window, start the second seed node with the following command:

mvn -pl killrweather exec:java -Dexec.args="2554"

2554 corresponds to the port of the second seed-nodes element in the configuration. In the log output you see that the cluster node has been started and joins the other seed node and becomes a member of the cluster. Its status changed to 'Up'. Switch over to the first terminal window and see in the log output that the member joined.

Some of the temperature aggregators that were originally on the ActorSystem on port 2553 will be migrated to the newly joined ActorSystem on port 2554. The migration is straightforward: the old actor is stopped and a fresh actor is started on the newly created ActorSystem. Notice this means the average is reset: if you want your state to be persisted you'll need to take care of this yourself. For this reason Cluster Sharding and Akka Persistence are such a popular combination.

Start another node in the third terminal window with the following command:

mvn -pl killrweather exec:java -Dexec.args="0"

Now you don't need to specify the port number, 0 means that it will use a random available port. It joins one of the configured seed nodes.
Look at the log output in the different terminal windows.

Start even more nodes in the same way, if you like.

Dynamic WeatherServer port

Each node's log will show its dynamic weather port opened for weather stations to connect to.

[2020-01-16 13:44:58,842] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-3] [] - WeatherServer online at http://127.0.0.1:12553/

Interacting with the HTTP endpoint manually Copy Section

With the cluster running you can interact with the HTTP endpoint using raw HTTP, for example with curl.

Record data for station 62:

curl -XPOST http://localhost:12553/weather/62 -H "Content-Type: application/json" --data '{"eventTime": 1579106781, "dataType": "temperature", "value": 10.3}'

Query average temperature for station 62:

curl "http://localhost:12553/weather/62?type=temperature&function=average"

The Fog Network Copy Section

In a new terminal start the Fog, (see Fog computing)

Each simulated remote weather station will attempt to connect to one of the round-robin assigned ports for Fog networking over HTTP.

The fog application, when run without parameters, will expect the cluster to have been started without parameters as well so that the HTTP ports if has bound are predictable.
If you have started cluster nodes manually providing port numbers you will have to do the same with the fog app or else it will not be able to find the endpoints.

For example:

mvn -pl killrweather exec:java -Dexec.args="2553"
sbt "killrweather-fog/runMain sample.killrweather.fog.Fog 8081 8033 8056"

Shutting down Copy Section

Shut down one of the nodes by pressing 'ctrl-c' in one of the terminal windows. The other nodes will detect the failure after a while, which you can see in the log output in the other terminals.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`samples/akka-sample-sharding-scala/README.md`) Copy Section

Cluster Sharding sample

The KillrWeather sample illustrates how to use Akka Cluster Sharding in Scala.
It also shows the basic usage of Akka HTTP.

To try this example locally, download the sources files with akka-sample-sharding-scala.zip.

The sample consists of two applications, each a separate maven submodule:

• killrweather - A distributed Akka cluster that shards weather stations, each keeping a set of recorded
  data points and allowing for local querying of the records. The cluster app has a HTTP endpoint for recording
  and querying the data per weather station.

• killrweather-fog - A client that periodically submits random weather measurements for a set of stations to a
  running cluster.

Let's start going through the implementation of the cluster!

KillrWeather Copy Chapter

Open KillrWeather.scala.
This program starts an ActorSystem which joins the cluster through configuration, and starts a Guardian actor for the system.

Guardian Copy Section

The Guardian.scala bootstraps the application to shard
WeatherStation actors across the cluster nodes.

Setting up sharding with the entity is done in WeatherStation.scala.
Keeping the setup logic together with the sharded actor and then calling it from the bootstrap logic of the application is a common pattern to structure sharded entities.

WeatherStation - sharded data by id Copy Section

Each sharded WeatherStation actor has a set of recorded data points for a station identifier.

It receives that data stream from remote devices via the HTTP endpoint. Each WeatherStation and also respond to queries about it's recorded set of data such as:

• current
• averages
• high/low

Receiving edge device data by data type Copy Section

The WeatherHttpServer is started with
WeatherRoutes
to receive and unmarshall data from remote devices by station ID to allow
querying. The HTTP port of each node is chosen from the port used for Akka Remoting plus 10000, so for a node running querying. To interact with the sharded entities it uses the EntityRef API.

The HTTP port of each node is chosen from the port used for Akka Remoting plus 10000, so for a node running
on port 2525 the HTTP port will be 12525.

Configuration Copy Section

This application is configured in killrweather/src/main/resources/application.conf
Before running, first make sure the correct settings are set for your system, as described in the akka-sample-cluster tutorial.

Fog Network Copy Chapter

Open Fog.scala.

Fog is the program simulating many weather stations and their devices which read and report data to clusters.
The name refers to Fog computing with edges - the remote weather station
nodes and their device edges.

This example starts simply with one actor per station and just reports one data type, temperature. In the wild, other devices would include:
pressure, precipitation, wind speed, wind direction, sky condition and dewpoint.
Fog starts the number of weather stations configured in killrweather-fog/src/main/resources/application.conf
upon boot.

Weather stations and devices Copy Section

Each WeatherStation is run on a task to trigger scheduled data sampling.
These samples are timestamped and sent to the cluster over HTTP using Akka HTTP.

Akka HTTP example Copy Chapter

Within KillrWeather are two simple sides to an HTTP equation.

Client

• WeatherStation - HTTP data marshall and send

Server

• WeatherHttpServer - HTTP server
• WeatherRoutes - HTTP routes receiver which will unmarshall and pass on the data

Both parts of the application uses Spray-JSON for marshalling and unmarshalling objects to and from JSON.

Running the samples Copy Chapter

The KillrWeather Cluster Copy Section

There are two ways to run the cluster, the first is a convenience quick start.

A simple three node cluster in the same JVM

The simplest way to run this sample is to run this in a terminal, if not already started:

sbt killrweather/run

This command starts three (the default) KillrWeather actor systems (a three node cluster) in the same JVM process.

Dynamic WeatherServer ports

In the log snippet below, note the dynamic weather ports opened by each KillrWeather node's WeatherServer for weather stations to connect to.
The number of ports are by default three, for the minimum three node cluster. You can start more cluster nodes, so these are dynamic to avoid bind errors.

[2020-01-16 13:44:58,842] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-3] [] - WeatherServer online at http://127.0.0.1:12553/
[2020-01-16 13:44:58,842] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-19] [] - WeatherServer online at http://127.0.0.1:53937/
[2020-01-16 13:44:58,843] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-15] [] - WeatherServer online at http://127.0.0.1:12554/

A three node cluster in separate JVMs

It is more interesting to run them in separate processes. Stop the application and then open three terminal windows.
In the first terminal window, start the first seed node with the following command:

sbt "killrweather/runMain sample.killrweather.KillrWeather 2553"

2553 corresponds to the port of the first seed-nodes element in the configuration. In the log output you see that the cluster node has been started and changed status to 'Up'.

You'll see a log message when a WeatherStation sends a message to record the current temperature, and for each of those you'll see a log message from the WeatherRoutes showing the action taken and the new average temperature.

In the second terminal window, start the second seed node with the following command:

sbt "killrweather/runMain sample.killrweather.KillrWeather 2554"

2554 corresponds to the port of the second seed-nodes element in the configuration. In the log output you see that the cluster node has been started and joins the other seed node and becomes a member of the cluster. Its status changed to 'Up'. Switch over to the first terminal window and see in the log output that the member joined.

Some of the temperature aggregators that were originally on the ActorSystem on port 2553 will be migrated to the newly joined ActorSystem on port 2554. The migration is straightforward: the old actor is stopped and a fresh actor is started on the newly created ActorSystem. Notice this means the average is reset: if you want your state to be persisted you'll need to take care of this yourself. For this reason Cluster Sharding and Akka Persistence are such a popular combination.

Start another node in the third terminal window with the following command:

sbt "killrweather/runMain sample.killrweather.KillrWeather 0"

Now you don't need to specify the port number, 0 means that it will use a random available port. It joins one of the configured seed nodes.
Look at the log output in the different terminal windows.

Start even more nodes in the same way, if you like.

Dynamic WeatherServer port

Each node's log will show its dynamic weather port opened for weather stations to connect to.

[2020-01-16 13:44:58,842] [INFO] [] [akka.actor.typed.ActorSystem] [KillrWeather-akka.actor.default-dispatcher-3] [] - WeatherServer online at http://127.0.0.1:12553/

Interacting with the HTTP endpoint manually Copy Section

With the cluster running you can interact with the HTTP endpoint using raw HTTP, for example with curl.

Record data for station 62:

curl -XPOST http://localhost:12553/weather/62 -H "Content-Type: application/json" --data '{"eventTime": 1579106781, "dataType": "temperature", "value": 10.3}'

Query average temperature for station 62:

curl "http://localhost:12553/weather/62?type=temperature&function=average"

The Fog Network Copy Section

In a new terminal start the Fog, (see Fog computing)

Each simulated remote weather station will attempt to connect to one of the round-robin assigned ports for Fog networking over HTTP.

The fog application, when run without parameters, will expect the cluster to have been started without parameters as well so that the HTTP ports if has bound are predictable.
If you have started cluster nodes manually providing port numbers you will have to do the same with the fog app or else it will not be able to find the endpoints.

For example:

sbt "killrweather-fog/runMain sample.killrweather.fog.Fog 8081 8033 8056"

Shutting down Copy Section

Shut down one of the nodes by pressing 'ctrl-c' in one of the terminal windows. The other nodes will detect the failure after a while, which you can see in the log output in the other terminals.

The Akka family of projects is managed by teams at Lightbend with help from the community.

License Copy Chapter

Akka is licensed under the Business Source License 1.1, please see the Akka License FAQ.

Overview (`akka-pki/src/test/resources/README.md`) Copy Section

Private keys for tests

Created using:

# Eliptic curve key
$ openssl ecparam -out ecdsa.pem -name secp256r1 -genkey
$ openssl pkcs8 -topk8 -nocrypt -in ecdsa.pem -out pkcs8-ecdsa.pem

Overview (`akka-remote/src/test/resources/ssl/README.md`) Copy Section

Testing Resources

This folder contains a diverse set of keys, certificates and keystores generated using scripts inspired by
those from https://github.com/playframework/play-samples/tree/2.8.x/play-scala-tls-example/scripts.

The resources in this folder are build with a few restrictions. See TlsResourcesSpec.scala for some
tests asserting the certificates and keys in this folder fulfill the required restrictions.

Overview (`akka-remote/src/test/resources/ssl/pem/README.md`) Copy Section

PEM resources

The following resources were copy/pasted from the akka-pki/src/test/resources folder.

• multi-prime-pkcs1.pem
• pkcs1.pem
• pkcs8.pem
• selfsigned-certificate.pem

Note that these set of certificates are valid PEMs but are invalid for akka-remote
use. Either the key length, certificate usage limitations (via the UsageKeyExtensions),
or the fact that the key's certificate is self-signed cause one of the following
errors: certificate_unknown, certificate verify message signature error/bad_certificate
during the SSLHandshake.

--- METRICS ---

• Files Extracted: 17
• Estimated Token Budget: ~21069 tokens
• Canonical Reference: https://codewiki.google/github.com/akka/akka-core