Saga Pattern with Kafka and NodeJS: simple implementation
Whenever you’ve decided to rebuild your application from monolithic to microservices architecture you come across with one major difficulty. This difficulty is to implement a distributed transaction for the operation which involves several microservices together.
Hopefully, there’re several patterns that may be applied. There’re at least 2 approaches available: Two-Phase Commit or Saga. Here we consider deeply Saga pattern only.
Let’s consider the canonical real example where we need to implement the distributed transaction. So we have some services: Flight Booking Service, Hotel Booking Service, and Payment Service. Each of those services has its own database and transaction boundaries within those DBs. We need to implement Trip Service which needs to book Flight, Hotel within one transaction.
Before we start to implement this, let’s have a look at the saga purpose. Sagas maintain data consistency across distributed services using a sequence of local transactions. Each local transaction updates data within a single service.
Any service initiates the first step of the saga. The completion of a local transaction triggers the execution of the next local transaction and so on.
If any issue happens during one of the transaction steps, for example, there are no available rooms or credit card riched their limits, then saga should be rolled back all local transactions which have been done before this step.
Suppose that the (n+1) transaction of a saga fails. The effect of the previous n transactions must be done. Conceptually. each of those steps, T(i) has a corresponding compensation transaction, C(i), which undoes the effect of the T(i). To undo the effect of those first n steps, the saga must execute each Ci in reverse order. The sequence of steps is T(1), T(2),…T(n), C(n), … C(1). In this example, T(n+1) fails, which requires steps T(1), T(2), … T(n) to be undone:
Well, this saga property provides:
- Atomicy for the transaction: all the changes are performed, or none of them are.
- Consistency: data is in a consistent state when a transaction starts and when it ends.
- Durability: changes to data persist and are not undone, even in the event of a system failure. Even when our Server Saga Orchestrator fails in the middle of a transaction, since we persist all steps in Kafka — as soon as Server spin up again it continues to handle step from place where it’s finished before failure.
But Saga is a lack Isolation property in ACID acronym since all intermediate commits are visible for other transactions while the transaction isn’t finished yet. But actually, it’s not an issue and we could overcome this by applying a bunch of countermeasures, we will consider them in the next article. But anyway, please keep in mind that Saga doesn’t provide Isolation.
Here is the article which provides examples of anomalies that may happen because of lack of Isolation and countermeasures which mitigate this:
Let’s have a look how this process may look like for a successful transaction:
For each saga step, we’re going to have a separate channel (Kafka Topic). 1 — Trip server starts saga invoking any external route of Saga Orchestarition Server and initiate saga transaction 2 — Saga Orchestration server posts a message to Kafka Topic which is responsible for the first saga step (Step1Channel) 3 — Saga Orchestrator Server subscribes to all channels for all saga steps and gets the first Message published in the previous step 4 — New got event from Step1Channel initiates invoking of external service (Flight Booking) to reserve a ticket 5 — FlighBooking Service returns success response that the ticket is reserved 6 — Saga Orchestrator posts event to Kafka topic belongs to saga step2 7–14 — the process for the rest steps is the same as discussed below
If for example, one step in the transaction fails, the workflow depicted here (the red line draws transaction compensation flow):
Let’s view how it would look like in a code to create a saga transaction:
As we discussed earlier each step has its own channel (topic) in Kafka: FlightBookingService, HotelBookingService, PaymentService.
In onReply method we implement any interaction with 3d party service, it may be anything. For example, we may invoke the REST API of Flight Booking Service.
Each transaction step has its own compensation action and is described in withCompensation method.
Let’s start to implement SagaDefinitionBuilder. It’s pretty straightforward, we just save metadata in private filed of SagaDefinitionBuilder instance:
Pay your attention to build method. Here we pass sagaDefinitions in SagaProcessor which is the heart of our Saga Orchestrator. For the implementation of SagaProcessor, we’re going to use the KafkaJS library for communication with Kafka:
When we initialize SagaProcessor for each step we create a separate channel (topic) and subscribe to all of them. We also have start method which initiates a saga transaction — it just posts the first message to Kafka. As soon as SagaProcessor gets a new message we handle all of them.
If invocation command described in onReply method successfully completed, SagaOrchestartor initiates the next step in Saga Transaction. If it fails — we chain all compensation actions in the reverse order.
This logs we will see if all steps completed in the transaction:
If for any reason the last server in the saga transaction isn’t available, we will get the following sequence of logs:
Full code provided here: https://github.com/vladimirtopolev/kafka-saga
Before running this application you need to spin up Kafka Cluster. The easiest way to do it — via Docker. Repository content docker-compose file in the root of application directory:
Run all necessary container just run this command from place where docker-compose file is:
docker-compose upThat’s it. I hope that this article was useful for you. Thank for your time.
