Message Queue Service using Kafka

Message Queue System using Kafka

In this article, we are going to create a Message Queue Service using Kafka and KafkaJS, somewhere similar to SQS, and we will make sure that messages are processed exactly once, in the exact order that they are sent. We will also implement the redelivery mechanism to make sure that if something fails in our business logic, we can add it to the queue again.

Find the source code on Github: https://github.com/icpsoni/kafka-message-queue

Prerequisites

1: Get Kafka

Download Kafka and extract it from here, and navigate to the directory.

$ tar -xzf <file_name>.tgz
$ cd <file_name>tec

2: Start the Kafka Environment

NOTE: Your local environment must have Java 8+ installed.

Run the following commands to start ZooKeeper:

# Start the ZooKeeper service
# Note: Soon, ZooKeeper will no longer be required by Apache Kafka.
$ bin/zookeeper-server-start.sh config/zookeeper.properties

Open another terminal session and run Kafka Broker service using:

# Start the Kafka broker service
$ bin/kafka-server-start.sh config/server.properties

Once all services have successfully launched, you will have a basic Kafka environment running and ready to use.

3: Create a Topic to Store our Queue Messages

With the following command, we will create a topic called “message-queue” in our Kafka.

$ bin/kafka-topics.sh --create --topic message-queue --bootstrap-server localhost:9092

To check the created topic use this command.

$ bin/kafka-topics.sh --describe --topic message-queue --bootstrap-server localhost:9092

You can create a topic with any number of partitions, In here we are using 1 partition only.

Creating Services Using KafkaJS

We basically need 3 things to make our queue system work perfectly.

  1. Sending Events to Kafka Queue topic.
  2. Subscribing to the topic and reading the Queue Message.
  3. Handling the redelivery.

kafka-config.js

The following snippet contains the basic config of Kafka that we need for our queue system.

// Using KafkaJs nodejs library
import { Kafka } from 'kafkajs';

// kafka broker running on localhost:9092 default port
const kafkaBroker = 'localhost:9092';

// kafka topic used for queue messages
export const kafkaTopic = 'message-queue';

// kafka client with basic config
export const KafkaClient = new Kafka({
brokers: [kafkaBroker]
});

producer.js

This service will send messages to the Kafka topic we created earlier. We can use sendMessageToQueue function and pass message object which needs to be sent to Kafka.

import { KafkaClient } from './index.js';
import { kafkaTopic } from "./kafka-config";

export const sendMessageToQueue = async (message) => {
const producer = KafkaClient.producer();
await producer.connect();
await producer.send({
topic: kafkaTopic,
messages: [
{
value: message // Your message data goes here
}
]
});
// Disconnect producer once message sending is done.
await producer.disconnect();
};

message-queue.js

This is our main service that takes care of receiving the messages from the Kafka queue doing the business logic and handling the redelivery if something goes wrong in business logic.

import { sendMessageToQueue } from "./producer.js";
import { KafkaClient, kafkaTopic, kafkaGroupId } from "./kafka-config.js";

export const consumeMessage = async () => {
// Creating a Consumer Instance
const consumer = KafkaClient.consumer({
groupId: kafkaGroupId,
});

await consumer.connect();
// Subscribing to out Kafka topic
await consumer.subscribe({ topic: kafkaTopic, fromBeginning: true});

await consumer.run({
autoCommit: false, // It won't commit message acknowledge to kafka until we don't do manually
eachMessage: async ({ topic, partition, message}) => {
const messageData = message.value.toString();
try {
// Do the business Logic
console.info('Received Message', messageData);
} catch (error) {
console.error(error);
// Resending message to kafka queue for redelivery
await sendMessageToQueue(messageData);
} finally {
const offset = +message.offset + 1;
// Committing the message offset to Kafka
await consumer.commitOffsets([{topic: kafkaTopic, partition, offset: offset.toString()}]);
}
}
});
};

Download the source

Bonus: It contains test sample files too.

References:

  1. https://kafka.apache.org/quickstart
  2. https://kafka.js.org/docs/getting-started

--

--

--

I’m Chandraprakash Soni, a full stack developer & engineer. Love to build robust and unit-tested web applications.

Love podcasts or audiobooks? Learn on the go with our new app.

Recommended from Medium

JavaScript Array Instance Methods that You May Have Missed

React, the Pros and the Cons

Picture of a wall of code because it’s a pretty picture to open a general programming blogpost

Mainichi: A Tale of React and Redux Pt. 1 — React

Efficient end-to-end testing for React-Native apps with Nix and Saucelabs

Why Do Developers Prefer Vue.Js Over Other Frameworks?

Customize your fonts with Cufon Font Replacement

//platform.twitter.com/widgets.js from Twitter https://twitter.com/AllMastersPlumb

React vs Vue — Components and Props Basics

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store
Chandraprakash Soni

Chandraprakash Soni

I’m Chandraprakash Soni, a full stack developer & engineer. Love to build robust and unit-tested web applications.

More from Medium

Manage a history of domain model changes with Spring

How to use Schema Registry Plugin with HTTPS enabled

Apache Kafka: A Basic Intro