Message Queue Service using Kafka

Chandraprakash Soni — Sun, 09 May 2021 19:18:00 +0000

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.

Sending Events to Kafka Queue topic.
Subscribing to the topic and reading the Queue Message.
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 the 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.
https://github.com/icpsoni/kafka-message-queue

References:

Getting Started with Caching (Redis + NodeJS)

Chandraprakash Soni — Tue, 04 Feb 2020 14:19:03 +0000

This post is a basic introduction to caching and how it works. We will be using Redis for caching with Nodejs to show you how we can take advantage of caching to improve the speed and performance of the application and to reduce the overhead from server resources.

What is Caching ?
Caching (pronounced “cashing”) is the process of storing data in a cache. A cache is a temporary storage area. A cache as a data store is easier for the client (or server) to reach, as opposed to a permanent data store that might be located on a different service, which takes more time and resources to reach (a database or an external API endpoint).

For example: The files you automatically request by looking at a Web page are stored on your hard disk in a cache subdirectory under the directory for your browser.

Now let’s take a look on Redis. (Source: Redis Official)

What is Redis ?
Redis is an open source (BSD licensed), in-memory data structure store, used as a database, cache and message broker. It supports data structures such as strings, hashes, lists, sets, sorted sets with range queries, bitmaps, hyperloglogs, geospatial indexes with radius queries and streams. Redis has built-in replication, Lua scripting, LRU eviction, transactions and different levels of on-disk persistence, and provides high availability via Redis Sentinel and automatic partitioning with Redis Cluster.

Let’s build very basic project to implement caching using redis:

DEV Community: Chandraprakash Soni

Message Queue Service using Kafka

Prerequisites

Creating Services Using KafkaJS

kafka-config.js

producer.js

message-queue.js

Getting Started with Caching (Redis + NodeJS)

Getting Started with Caching (Redis + NodeJS) | by Chandraprakash Soni | Medium

Chandraprakash Soni ・ Dec 3, 2019 ・
Medium

DEV Community: Chandraprakash Soni

Message Queue Service using Kafka

Prerequisites

Creating Services Using KafkaJS

kafka-config.js

producer.js

message-queue.js

Getting Started with Caching (Redis + NodeJS)

Getting Started with Caching (Redis + NodeJS) | by Chandraprakash Soni | Medium

Chandraprakash Soni ・ Dec 3, 2019 ・ Medium

Chandraprakash Soni ・ Dec 3, 2019 ・
Medium