Jai Kisan Kafka Package (jk-kafka)

Jai Kisan Kafka Package is a simple and intuitive package that allows you to easily consume and produce messages in AWS MSK (Managed Streaming for Apache Kafka) and validate them against a JSON schema.

Installation

To install jk-kaka package, you can use npm:

npm install jk-kafka

Define env file

To use package we need to define .env file in root . Here is the sample file for env

BROKERS=broker1:9092,broker2:9092
CLIENT_ID=kafka-producer
SCHEMA_REGISTRY_HOST=http://schema-registry:8081
PRODUCER_CLIENT_ID=kafka
PRODUCER_METADATA_BROKER_LIST=broker1:9092,broker2:9092
CONSUMER_GROUP_ID=my-group
CONSUMER_METADATA_BROKER_LIST=broker1:9092,broker2:9092

Producer

To use the producer, you can import it and call the SendMessage function. Here's an example:

const { kafkaProducer } = require('jk-kafka');
const producerExec = async () => {
const requestBody = {
      topic: 'my-topic-1',
      message: {
        id: 1,
        name: 'Jai-Kisan',
        email: '[email protected]',

      },
      partition: 0,
      isValidationEnabled: false,
      key: 'my-key',
      schemaId: 22,
    }


    await kafkaProducer.connect()
    const result = await kafkaProducer.sendMessage(requestBody);
    console.log(result);
    await kafkaProducer.disconnect()
}
producerExec()

In the example above, we first import the SendMessage function from jk-kafka-beta. We then define the topic, message, partition, isValidationEnabled, key, and schemaId variables, and call the SendMessage function with object as a parameter. The function sends the message to the specified topic, and returns an object with the message and the response from the producer.Finally, the producer is disconnected using the disconnect function to free up resources.

Consumer

To use the general consumer, you can import it and call the kafkaConsumer function. Here's an example:

const { kafkaConsumer } = require('jk-kafka');

const config = {
// Define the onMessage callback function
  onMessage: (message) => {
    console.log(`Received message: ${JSON.stringify(message)}`);
  }
};

const subscribe = {
  topic:'my-topic-1'
}

const consumerExec = async () => {
try{
await kafkaConsumer.connect()
await kafkaConsumer.subscribe(subscribe)
await kafkaConsumer.consumeMessage(config)
}
catch(error){
  console.error(error)
} finally {
  await kafkaConsumer.disconnect();
}
}

consumerExec(); 

In the example above, we first import the kafkaConsumer function from jk-kafka. We then call the connect method to connect with broker and subscribe method to subscribe topic. The consumeMessage function calls with predefined onmessage functions which execute on each message received.

Consumer (Lambda)

To use the general consumer, you can import it and call the kafkaLambdaConsumer function. Here's an example:

const { kafkaLambdaConsumer } = require('jk-kafka');

exports.handler = async (event, context) => {
  // Define the onMessage callback function
  const config = {
    event: event,
    onMessage: (message) => {
      console.log(`Received message: ${JSON.stringify(message)}`);
    }
  };
  
  try {
    await kafkaLambdaConsumer.consumeMessage(config);
  } catch (error) {
    console.error(error);
  }
};

In the example above, we first import the kafkaLambdaConsumer function from jk-kafka. Lambda act as consumer which will will be configured via MSK trigger(Consumer group and topic we can define on lambda-msk trigger configuration). The consumeMessage function calls with predefined onmessage functions and incoming event which execute on each message received.

Contributing

Contributions are welcome! If you have any issues or feature requests, please open an issue on the GitHub repository. If you'd like to contribute code, please fork the repository and submit a pull request.

Local Setup

In order to contribute to the development of a Node.js project or run it locally, you will typically need to install the required npm packages.

npm install --save

Run Test

ts-node producer.ts
ts-node consumer.ts

License

Jai Kisan Kafka Package is MIT licensed.