In part 5 of this series, we started to scale the application from a local one to a clustered one by introducing Cluster Sharding. In this article, we will continue our effort to scale the payment processor out and make it resilient through the use of Cluster Singletons and Cluster-Aware Routers.
Before we get started, here’s a quick reminder of what we’ve seen so far: in the first part we had a look at the the raison d’être of Akka Typed and what advantages it has over the classic, untyped API. In the second part we introduced the concepts of message adapters, the ask pattern and actor discovery. In the third part, we covered one of the core concepts of Actor systems: supervision and failure recovery. In the fourth part, we covered one of the most popular use-cases for Akka: event sourcing. Finally, as mentioned before, in the fifth part we started scaling out by making use of cluster sharding.
As a reminder, this is where we want to get to:
We started to distribute the typed payment processor on many nodes using Akka Cluster in the previous article, but are not finished yet. There are two things we need to still be taking care of:
- make sure that there is only one
Configurationactor in the cluster – or else we’d be risking to run multiple versions of this persistent actor at the same time, which is a big no-go - scale out the
CreditCardProcessoractors in order to increase throughput (by having many of them) and availability (by distributing them on several nodes)
Let’s get started with deploying the Configuration actor as a so-called Cluster Singleton.
Cluster Singleton
The Cluster Singleton feature of Akka Cluster takes care of managing the deployment of an actor in the cluster so that there’s only one instance active at all times. In order to do so, it:
- runs the singleton actor on the oldest node of the cluster
- restarts the actor on another node if the oldest node becomes unavailable
- takes care of buffering messages so long as there is no instance of the cluster singleton running
So what does it take to make use of this feature? With the new Akka Typed API, it is rather straightforward. Previously, we were just starting the configuration actor as a child of the PaymentProcessorguardian:
|
1 |
val configuration = context.spawn(Configuration(), "config") |
In order to make use of cluster sharding, we only need to use the ClusterShardingextension, like so:
|
1 2 3 4 5 6 |
import akka.cluster.typed.{ClusterSingleton, SingletonActor} val configuration: ActorRef[Configuration.ConfigurationRequest] = ClusterSingleton(context.system).init(SingletonActor( Configuration(), "config" )) |
If you are familiar with the Classic version, you’ll appreciate how much simpler things got – there’s no more need to configure a separate ClusterSingletonManagerand ClusterSingletonProxy, Akka takes care of this. The returned ActorRef[M]is the one you can use in order to send messages to the singleton, and should you not have one at disposal, it is okay to run the initmethod multiple times.
Note that you should be careful to not make too extensive use of cluster singletons in your system – they are, after all, prone to becoming a bottleneck (as there’s only one of them). Make sure you read the documentation to be aware of the trade-offs you are making.
Cluster Aware Routers
There’s one last component in our system that needs some attention with the move to a clustered environment: the CreditCardProcessor. There are two issues with it:
- it will become a bottleneck as there is only one instance of it running per cluster node
- it is backed by a persistent storage implementation that is started on each node, yet uses Akka persistence in the background with the same persistent identifier, which is a big no-go
As part of this article, we’ll fix both these issues but leave room for optimization for the second issue (and thus, room for one article on the topic).
Let’s get started by increasing the amount of CreditCardProcessor-s available in our cluster. Up until now, we only had created one instance of a processor per node. Let’s change this a bit and start 10 of them per node in the PaymentProcessorguardian:
|
1 2 3 4 5 6 7 8 |
val supervisedCreditCardProcessor = Behaviors .supervise(CreditCardProcessor()) .onFailure[Exception]( SupervisorStrategy.restartWithBackoff(minBackoff = 5.seconds, maxBackoff = 1.minute, randomFactor = 0.2)) for (i <- 1 to 10) { context.spawn(supervisedCreditCardProcessor, s"creditCardProcessor-<strong>$</strong>i") } |
Note that we are still using a custom supervision strategy that will restart the CreditCardProcessoractors (with backoff) instead of stopping them – more on this later.
In the second article of this series, we have setup the CreditCardProcessorto register itself with the Receptionist. This is going to be quite useful now as we’ll leverage the fact that the Receptionistis cluster-aware. Instead of looking up credit card processor’s individually in PaymentHandling using the ServiceKey, we’ll leave this to a new group router which we’ll pass to the sharded PaymentRequestHandler-s (as they are the ones that need to interact with the processors).
|
1 2 3 4 5 6 7 |
import akka.actor.typed.scaladsl._ // initialize the group router for credit card processors val creditCardProcessorRouter = context.spawn( Routers.group(CreditCardProcessor.Key), "creditCardProcessors" ) |
And that’s pretty much all we need to create a new router which we can then pass to the PaymentRequestHandlerinstead of the raw Listingreturned by the receptionist. Note that the router doesn’t supervise the CreditCardProcessor routees in any way, which is where our custom supervision comes in to make sure that the 10 instances keep alive over time in case of crash.
Finally, we need to make sure that we don’t create multiple instance of the CreditCardStorageimplementation. In this article and as a first solution, we will just do so by deploying it as a cluster singleton – but this part is left to the reader (or you can always check the sources that go with this article).
And in the next article of this series, we’ll improve the credit card storage by looking into the Typed version of Distributed Data. Stay tuned!
Concept comparison table
As usual in this series, here’s an attempt at comparing concepts in Akka Classic and Akka Typed (see also the official learning guide):
ClusterSingletonManagerand ClusterSingltonProxy | ClusterSingleton(system).init(...) |
| Cluster Aware Pool Routers | – |
| Cluster Aware Group Routers | Register routees to the Receptionistwith their service key and then get a group router using the service key with Routers.group(CreditCardProcessor.Keey) |
Comments 8
I have a question on how to interact with this actor system.
I see that you have
ActorSystem[Nothing](PaymentProcessor(), "typed-payment-processor")inMain.scala. But how can I get the reference of thePaymentHandlingactor and send messages to it?Author
Accessing it isn’t yet part of the example, because I haven’t developed it until then. One way to go could be for example to provide the actor reference of
PaymentHandlingto clients subscribing toPaymentProcessorfrom the outside — but in my mind,PaymentProcessorwould also spawn an API that lets you interact with the handlingHello,
This has been an excellent series. I have two questions related to this post and the last, will be obliged if you could please answer.
I have a cluster where I have persistent entities. HTTP requests can come to any node in the cluster. Since for a given entity Id, we can only have one actor instance, I made that a singleton. However, the entity actor only replies if the HTTP request comes to the node that hosts the singleton. On all other nodes, I get an ask timeout. My entity commands do have a replyTo, hence I used the ask pattern.
Then I changed it to use cluster sharding. However, if I ask by getting the entity reference from the Sharding system, I get the same behaviour as above. I believe I need to ask the sharding region. However, I am able to tell the ShardingRegion by wrapping the command in the ShardingEnvelope. I am not sure, how to wrap the command in an envelope to do an ask.
Many Thanks
Author
Hi Alan,
sorry for the late reply. It seems that in your use of singleton, you may have directed the messages to the wrong address. Instead of using an address directly, you need to retrieve the reference to the singleton using the
ClusterSingleton.initmethod (as described in the documentation). For sharding, it is more of a matter of configuring the correct message extractor ID – then you don’t need a special envelope, you can just use your domain messages as they are. See this article about cluster sharding)This article is very helpful. Thank you very much Manuel. Is there any way to dynamically resize routers in Typed?
Author
Hi Raj,
there currently isn’t a way to do this in Typed. You could skip using routers and create your own parent actor that dynamically allocates more child actors if really necessary. That being said, I seldomly encountered the need to do this for routers on a single machine. For multiple machines, adding one more machine / node should allow to automatically add the routees to the group. Hope this helps.
Hi.
If we are maintaining some internal data inside our ClusterSingleton(job’s queue for example), what is better way to replicate this data across cluster nodes? We do not want to lose any job received from external system, if current singleton node goes down. What’s correct way to do it?
Author
Hi Ivan,
there are several possibilities to make sure that your data can survive the crash of the node that currently holds the singleton:
– use Akka Persistence to store the data (in the case of a job queue, each added job would be an event)
– use Akka Distributed Data to replicate the data on each node – in this case you’ll need to fit the data in the shape of the predefined CvRDTs (ORSet, ORMap, etc.)
– use any other persistence mechanism at your disposal, such as an existing RDBMS, directly. As long as you can ensure that each node will have access to it and that it provides replication on its own (e.g. Cassandra, or a replicated Postgres / or if you’re on AWS, AWS Aurora) that’s a viable solution as well