The repo consists of two branches:

The main task is to create two separate microservices that communicate via Kafka. One of these services exposes REST endpoints and produce messages to the Kafka broker. The other service consumes these messages and write them to an in-memory storage. Both services are maintained in a Monorepo.

This branch has additional modules under each microservice trying to tackle idempotency at api-service(a.k.a producer) and database-service (a.k.a consumer) services.

Assuming kafka broker is available at this address, as configured at ./api-service/src/config/configuration.ts

 brokers: 'localhost:9092'

start services locally

  git clone https://github.com/maheshKRex/nestjs-kafka-microservice.git
  cd nestjs-kafka-microservice
  cd api-service
  npm install
  npm run start:dev

  cd ../database-service
  npm install
  npm run start:dev
  

To run tests, run the following command

  cd api-service
  npm run test:e2e

POST /user

curl --location 'http://localhost:3001/users' \
--header 'Content-Type: application/json' \
--data '{
    "id":"1",
    "name": "test user 1"
}'

GET /user/1

curl --location 'http://localhost:3001/users/1'

{
    "id": "1",
    "name": "test user 1"
}

Added a way to track processed messages by a unique messageId. Each message needs to have a unique messageId assigned by the producing service, as a Kafka message header. When a new message is consumed the in-memory array is checked for the existence of this message Id. If present, then the message is a duplicate. The consumer updates its offsets to effectively mark the message as consumed to ensure it is not redelivered, and no further action takes place.

git checkout idempotent-solution
./api-service/npm run start:dev
./database-service/npm run start:dev

create user with messageId 1111 POST /user

curl --location 'http://localhost:3001/users' \
--header 'Content-Type: application/json' \
--data '{
    "id":"1",
    "messageId":"1111",
    "name": "test user 1"
}'

retry the same curl command by sending the same value as the messageId 1111 POST /user

curl --location 'http://localhost:3001/users' \
--header 'Content-Type: application/json' \
--data '{
    "id":"1",
    "messageId":"1111",
    "name": "test user 1"
}'

check consumer console for detection of the duplicate message and the consumer commits the offset to avoid further duplications by other consumers.

https://ibb.co/GRbKXcZ

More Optimizations can be done on idempotent consumer solution explored here.

• For instance, the logic of storing processed messageId and actuallty processing a message can be made atomic using transactions. There by avoid chances of failing to mark a message as processed or marking a message as processed and crashing the consumer, without really processing the message.

generally on the whole project:

• can introduce Mono-repo management tools like nx comes with easy managemet tools: https://nx.dev/
• shared settings between producer and consumer such as kafka broker addresses, topics can be moved to a shared library, to avoid duplicate code, there by not having to update settings at both services.