Akka and Kotlin coroutines: ♡

I’ve experimented with Kotlin and coroutines in programming Akka. And I must say, I really like the combination so far. But before I go into it some brief preliminaries for those who don’t know Akka and actors.

Actors and Akka

Actors are a programming model that fits cloud native architectures particularly well. Being highly available and scaling horizontally. All while embracing the realities of multiple servers collaborating, server instances coming and going and the network having hickups.

On the JVM Akka is the prominent actor framework. It’s been around for a while now and as a result it’s highly reliable, well thought out and offers a wide programming eco system. My own interest in Akka is because of its suitability for software systems that can only be built with business events as a key construct and thinking model.

And then of course materialized views, CQRS and near real-time data streams play a big role in constructing those systems.

Akka and Java programming; the challenge

Akka’s Java api can be a bit unwieldy I find. Which is to be expected since Java as a programming language is still lagging quite a bit compared to newer programming languages.

And because Akka actors have to be non-blocking in their implementation. Which means extra bookkeeping we have to do to maintain intermediate state that would otherwise be handled by the JVM stack.

So I thought I’d see what Kotlin and coroutines can do to improve the programming experience. Trying to make the programming model feel as much native Kotlin coroutines as I can.

Akka and Kotlin coroutines demo

Our example actor receives requests with a sequence number, keeps a counter of processed messages, does a call to a slow service and sends a response using this data. For this example the actor has to process messages sequentially.

During the call to `slowService` we can’t block the thread because that would make the actor unresponsive to system messages. Instead with the Java api we’d have to kick off the Java `CompletableFuture`, make sure the result when it becomes available is sent to ourselves, go to a waiting mode where we stash all incoming messages for later processing, and when we do receive the result we get our stored message, send the end result and go back to processing mode.

All perfectly doable but the required call stack bookkeeping obscures the high level intent of what we’re trying to do.

By contrast using Kotlin suspend functions and coroutines the functionality of our actor is just as easy to read as blocking code. The message that we’re currently processing is still in scope after the service call so we can just use it and service response to return a result. But it’s still _non blocking_ code. So how does that work?

The `ChannelActor` is an Akka actor that is a `CoroutineScope`. So that we can launch a coroutine (think of it as a lightweight thread) in our actor and be confident that coroutine is stopped when the actor is stopped.

When the actor is started with `startReceive` we kick off a coroutine to start processing messages. IntelliJ shows next to `startReceive` that we get a `this: ActorScope` to work with. As a result we have a `ReceiveChannel` that’s the inbox of our actor. (See below for more information on the signatures.) We can maintain mutable state `counter` since we have one single coroutine doing the processing. With our `for` statement we’re looping over a suspending iterator. To the left you see a a special arrow icon in IntelliJ that indicates that our code will suspend there. Only to continue when we receive a message. We see the same icon next to the call to slowService. So the processing will suspend until we get a response. And then the actor can return the result. Suspending until the next message comes in.

But what if our `slowService` call is _too_ slow? Of course we configured our underlying http client and circuitbreaker with a timeout. Our `slowService` suspend function will throw an exception in that case. In our current actor that exception is unhandled and our processing of messages will be aborted. We decide that we want to return an exception message to our client in that case and continue processing.

We can use standard `try` `catch` exception handling for this.

Running our actor

So how do we run our demo actor and see that it works?

For testing purposes with runAkka we get an Akka actor system in scope and we’re in a coroutine scope so we can call suspend functions and launch coroutines. With the extension function `demoActor` we get a reference to our actor. Next to our call `channelActor.getCounter` we see the icon which tells us our code will suspend there until we get a result or it times out.

We use Kotlin extension functions to document the protocol of our demo actor.


Personally I really like how well Akka actors and Kotlin coroutines work together. It makes for a very accessible programming model I find. This demo uses Akka classic actors. It’s programming model is a natural fit with Kotlin coroutine channels. I’m still chewing on what a Kotlin coroutine version of Akka typed would look like. In a follow up post I’ll go into the design principles of Kotlin and coroutines they work out in this api.

PS: actors

You may wonder why go to all this trouble when there are already [actors available](https://kotlinlang.org/docs/shared-mutable-state-and-concurrency.html#actors) as part of Kotlin coroutines. The difference is that Kotlin coroutine actors are a great way to manage mutable state in the face of concurrency.

But they are confined to the memory of one process. They aren’t remote addressable, do not survive a server rebooting and can’t collaborate across server instances.

Akka actors on the other hand can live in a cluster of servers and thus be highly available and scale horizontally. They can be persistent and thus are very suitable for CQRS.

PPS: some supporting information

The `this: ActorScope` type is as follows

where `ReceiveChannel` is a standard Kotlin receive channel. `ActorScope` extends Kotlin `CoroutineScope` so that we can safely launch coroutines.

The `startReceive` method has the following signature

It uses Kotlin receiver functions which are idiomatic in Kotlin coroutine programming. The block is suspending so we can call suspending functions like `slowService`.

Another way of handling the messages would be

The signature of the `runAkka` method is as follows

Where `AkkaScope` extends `CoroutineScope` to allow launching coroutines and provides the actor system

This article was first posted on github: https://gist.github.com/joost-de-vries/479bbd285c4de23cb5bade21d83e2e22

This entry was posted in Blog, Kotlin. Bookmark the permalink.

Leave a Reply

Your email address will not be published. Required fields are marked *