Using Apache Kafka to implement event-driven microservices
Tags: kafka, eventdriven, microservices, java
Published: 2019-08-23 22:03:23

When talking about microservices architecture, most people think of a network of stateless services which communicate through HTTP (one may call it RESTful or not, depending on how much of a nitpicker one is).   
But there is another way, which may be more suitable depending on the use case at hand.  
I am talking about event-driven microservices, where in addition to the classic request-response pattern, services publish messages which represent events (facts) and subscribe to topics (or queues depending on the terminology used) to receive events/messages.  
To fully understand and embrace this new software design paradigm is not straight-forward but it is totally worth it (at least looking into it).
There are several interconnected concepts which need to be explored in order to discover the advantages of event-driven design and the evolutionary path which led to it, for example:
* Log (including log-structured storage engine and write-ahead log)
* Materialized View
* Event Sourcing
* Command Query Responsibility Segregation (CQRS)
* Stream processing
* "_Inside-out_" databases (a.k.a. "_un-bundled_" databases)

I'd like to point you to the following books to get familiar with those topics:

* [**Designing Data-Intensive Applications**](https://amzn.to/31S2Trb) by Martin Kleppmann - this is a very comprehensive book, it starts covering single-node application concepts, then distributed systems and finally batch and stream processing.
* [**Designing Event-Driven Systems**](https://www.confluent.io/designing-event-driven-systems) by Ben Stopford - it focuses on Apache Kafka as the backbone of event-driven systems.
* [**Kafka: The Definitive Guide**](https://amzn.to/2KDRqWc) by Neha Narkhede, Gwen Shapira, et al. - a more practical book than the other two.

I read those three books and then I started building a simple PoC since learning new design ideas is great but it is not completed until you also put them in practice.  
Also, I was not happy with the examples of event-driven applications/services available online, I found them too simplistic and not properly explained, so I decided to create one.  

### The proof of concept
The source code is split in two GitHub repositories (as per [the Clean Architecture](https://blog.cleancoder.com/uncle-bob/2012/08/13/the-clean-architecture.html)):

* [transfers_api](https://github.com/VictorGil/transfers_api) → contains Java entities and Avro data definition files
* [transfers_recording_service] (https://github.com/VictorGil/transfers_recording_service) → contains the business logic and the Kafka-related code

The proof of concept service keeps track of the balance available in bank accounts (like a ledger).
It listens for [_Transfer_](https://github.com/VictorGil/transfers_api/blob/1e7e4519dce6a6d1bf7628738e3358490c706d6f/src/main/resources/avro/transfer.avsc) messages on a Kafka topic and when one is received, it updates the balance of the related account by publishing a new [_AccountBalance_](https://github.com/VictorGil/transfers_api/blob/1e7e4519dce6a6d1bf7628738e3358490c706d6f/src/main/resources/avro/accountBalance.avsc) message on another Kafka topic.  

Please note that each entity type is represented by two different classes: 

* one is generated by Apache Avro and it's used for serialization and deserialization (so they can be sent and received from Kafka) → see [avro](https://github.com/VictorGil/transfers_api/tree/1e7e4519dce6a6d1bf7628738e3358490c706d6f/src/main/resources/avro) directory.
* the other one is a POJO which may contain some convenience constructors and does not depend on Avro → see [net.devaction.entity](https://github.com/VictorGil/transfers_api/blob/1e7e4519dce6a6d1bf7628738e3358490c706d6f/src/main/java/net/devaction/entity) package.

The [net.devaction.kafka.avro.util](https://github.com/VictorGil/transfers_api/tree/1e7e4519dce6a6d1bf7628738e3358490c706d6f/src/main/java/net/devaction/kafka/avro/util) package holds converters to move back and forth from one data representation to the other.  
In the beginning, Apache Kafka may seem overwhelming, even though it resembles a classic messaging broker such as ActiveMQ or RabbitMQ, it is much more than that and it works very differently internally.
Also, there are several Kafka client APIs, which adds more confusion to the learner.
We are going to focus on the following three:

* The [Producer API](https://kafka.apache.org/documentation/#producerapi) and the [Consumer API](https://kafka.apache.org/documentation/#consumerapi)
* The [Streams API](https://kafka.apache.org/documentation/#streamsapi)

The Producer and the Consumer APIs are lower level and the Streams API is built on top of them.
Both sets of APIs have advantages and disadvantages.
The Producer/Consumer API provides finer control to the application developer at the cost of higher complexity.
On the other hand, the Streams API is not as flexible but it allows the implementation of some standard operations more easily and it requires much less code.

The "**transfers recording**" example/PoC service can be started in one of the following two modes:

* The (explicit) **polling** mode which uses the Producer API, the Consumer API and also the Streams API (to create a queryable materialized view backed by a local data store, more on that later) → see [TransfersRecordingServicePolling.java](https://github.com/VictorGil/transfers_recording_service/blob/eecd02b7f893afcb7d96a32907a968713b7534ca/src/main/java/net/devaction/kafka/transfersrecordingservice/main/TransfersRecordingServicePolling.java)
* The "**join streams**" mode which only uses the Streams API → see [TransfersRecordingServiceJoinStreams.java](https://github.com/VictorGil/transfers_recording_service/blob/eecd02b7f893afcb7d96a32907a968713b7534ca/src/main/java/net/devaction/kafka/transfersrecordingservice/joinstreams/TransfersRecordingServiceJoinStreams.java)

The two modes provide exactly the same functionality which is quite convenient for comparison purposes.

### The (explicit) polling mode

It has four main components:

* A [_consumer_](https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html) which listens on the "transfers" topic → see [TransferConsumer.java](https://github.com/VictorGil/transfers_recording_service/blob/2b5cfcc6544fa9dc7611f2ca29dac7f765e395c5/src/main/java/net/devaction/kafka/transfersrecordingservice/transferconsumer/TransferConsumer.java#L103)
* A [ReadOnlyKeyValueStore](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.html) (which is part of the Streams API) to materialized the "account-balances" topic data into a queryable view, so we can use the accountId value to retrieve the latest/current balance of a specific account → see [AccountBalanceRetrieverImpl.java](https://github.com/VictorGil/transfers_recording_service/blob/eecd02b7f893afcb7d96a32907a968713b7534ca/src/main/java/net/devaction/kafka/transfersrecordingservice/accountbalanceproducer/AccountBalanceRetrieverImpl.java#L93).
Please note that the _accountId_ value is extracted from the "_transfer_" data message received by the consumer.
* The business logic which creates a new/updated _AccountBalanceEntity_ object from the received _TransferEntity_ object and the current _AccountBalanceEntity_ present in Kafka → see [NewAccountBalanceProvider.java](https://github.com/VictorGil/transfers_recording_service/blob/eecd02b7f893afcb7d96a32907a968713b7534ca/src/main/java/net/devaction/kafka/transfersrecordingservice/core/NewAccountBalanceProvider.java#L15)
* A [_producer_](https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html) which publishes the updated balance by sending a message to the "_account-balances_" topic → and the local data store will get updated accordingly.

### The "join streams" mode
As we said before, this second operating mode only uses the Streams API/DSL and taking advantage of it, we can code at a higher level of abstraction:

* Instead of a [_consumer_](https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html), we use a [KStream](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KStream.html) → see [TransfersRecordingServiceJoinStreams.java](https://github.com/VictorGil/transfers_recording_service/blob/271f1e7716b0298e18b90578eed6d978e0cd1d5f/src/main/java/net/devaction/kafka/transfersrecordingservice/joinstreams/TransfersRecordingServiceJoinStreams.java#L62)
* Instead of a [ReadOnlyKeyValueStore](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.html), we use a [KTable](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KTable.html) → see [TransfersRecordingServiceJoinStreams.java](https://github.com/VictorGil/transfers_recording_service/blob/271f1e7716b0298e18b90578eed6d978e0cd1d5f/src/main/java/net/devaction/kafka/transfersrecordingservice/joinstreams/TransfersRecordingServiceJoinStreams.java#L69) again.
* The business logic remains the same but now we call it from the [ValueJoiner](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/ValueJoiner.html) → see [TransferAndAccountBalanceJoiner.java](https://github.com/VictorGil/transfers_recording_service/blob/eecd02b7f893afcb7d96a32907a968713b7534ca/src/main/java/net/devaction/kafka/transfersrecordingservice/joinstreams/TransferAndAccountBalanceJoiner.java#L40)
* Instead of a [_producer_](https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html), we redirect the output of the streams join to the desired topic (i.e., "account-balances") → see [TransfersRecordingServiceJoinStreams.java](https://github.com/VictorGil/transfers_recording_service/blob/271f1e7716b0298e18b90578eed6d978e0cd1d5f/src/main/java/net/devaction/kafka/transfersrecordingservice/joinstreams/TransfersRecordingServiceJoinStreams.java#L75) once more → and the KTable will get updated with each message, closing the loop.

We can see that the code is much more compact than in the previous mode.
We do not need to explicitly map the _KStream_ key to the _KTable_ key, that's exactly what the _join_ does (see line 11 in the code snippet below). Hence, we need to choose the Kafka keys accordingly. In this case, both message keys represent the _account_ _id_.

{% gist https://gist.github.com/VictorGil/118ae1a7605bacdb18012056d4af6050 %}

### Diagram
<figure>
  <img src="https://thepracticaldev.s3.amazonaws.com/i/9j4pz1902xmjxdagqp4o.png" alt="Image depicting the diagram of the example/PoC Kafka Streams application">
  <figcaption>Diagram of the example/PoC Kafka Streams application</figcaption>
</figure>

### Running the code
To build and run the PoC application, in addition to Maven and Java, we also need a Kafka broker.
I decided to install the [Confluent Platform](https://www.confluent.io/product/confluent-platform/) which includes a Kafka broker (or a cluster depending on the configuration chosen) with some example topics and pre-configured integration with ElasticSearch and Kibana. But more importantly, it also includes an admin Web UI called _Control_ _Center_ which comes in very handy.  

I hit a few bumps when running the Confluent Platform for the first time on my Fedora 30 computer.
Namely, I had to manually install a couple of software packages (i.e., "_jot_" and "_jq_").
And I had to separately install the [Confluent CLI](https://docs.confluent.io/current/cli/index.html).
I also had to perform some several changes to some properties files and bash scripts to be able to run the Confluent Platform using a non-root user, [here](https://gist.github.com/VictorGil/5866f90dc626126897a76dbee53deeec) are the changes, please modify the configuration values as per your environment. 

Watch the following [YouTube video](https://youtu.be/w-Vy6_0buYo) to get all the details including starting the Confluent Platform and running the example Streams application:
{% youtube w-Vy6_0buYo %}