Kafka, AVRO and TypeScript?

Eduard Gert — Fri, 05 Mar 2021 08:09:56 +0000

In this article I want to show a simple example of how you can produce and consume Kafka messages with the AVRO format using TypeScript/JavaScript and KafkaJS.

What is Kafka?

Apache Kafka is a very popular event streaming platform and used in a lot of companies right now. If you want to learn more about Kafka, check out the official website.

However, since the whole ecosystem is based on JVM (Java, Scala, Kotlin), I never really checked for clients in other languages.

Recently I was playing around with a project in TypeScript and since it would have been handy to stream the results directly into Kafka, I checked for a JavaScript client and found KafkaJS. And it even plays well with AVRO.

How to use it?

Here is a simple example for an AVRO producer and consumer.

Set up a new node project and install these two dependencies. The schema registry is required to work with AVRO schemas.

npm install kafkajs @kafkajs/confluent-schema-registry

Configuring the Kafka connection

This example is in TypeScript but in JS it would work more or less in a similar way.
First import all the dependencies and configure all Kafka related settings.

import { Kafka } from "kafkajs";
import {
  SchemaRegistry,
  readAVSCAsync,
} from "@kafkajs/confluent-schema-registry";

const TOPIC = "my_topic";

// configure Kafka broker
const kafka = new Kafka({
  clientId: "some-client-id",
  brokers: ["localhost:29092"],
});

// If we use AVRO, we need to configure a Schema Registry
// which keeps track of the schema
const registry = new SchemaRegistry({
  host: "http://localhost:8085",
});

// create a producer which will be used for producing messages
const producer = kafka.producer();

const consumer = kafka.consumer({
  groupId: "group_id_1",
});

// declaring a TypeScript type for our message structure
declare type MyMessage = {
  id: string;
  value: number;
};

Create an AVRO schema

Now we need to make sure we can encode messages in AVRO. Therefore we need to be able to read a schema from a file and register it in the schema registry.

This is how the schema in this example will look like. Pretty straightforward, two fields called id which is a string and value which is an integer.
Insert this to a file called schema.avsc, we will use the confluent-schema-registry package to read it and register the schema in the schema registry.

{
  "name": "example",
  "type": "record",
  "namespace": "com.my.company",
  "doc": "Kafka JS example schema",
  "fields": [
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "value",
      "type": "int"
    }
  ]
}

Register an AVRO schema

Here is the function which we will use to read an AVRO schema from a file and register it in the schema registry.

// This will create an AVRO schema from an .avsc file
const registerSchema = async () => {
  try {
    const schema = await readAVSCAsync("./avro/schema.avsc");
    const { id } = await registry.register(schema);
    return id;
  } catch (e) {
    console.log(e);
  }
};

Produce a message using the AVRO schema

This is how we can build a producer. Before pushing a message (of type MyMessage which we defined above) we will encode it using the AVRO schema from the registry.

// push the actual message to kafka
const produceToKafka = async (registryId: number, message: MyMessage) => {
  await producer.connect();

  // compose the message: the key is a string
  // the value will be encoded using the avro schema
  const outgoingMessage = {
    key: message.id,
    value: await registry.encode(registryId, message),
  };

  // send the message to the previously created topic
  await producer.send({
    topic: TOPIC,
    messages: [outgoingMessage],
  });

  // disconnect the producer
  await producer.disconnect();
};

Create a Kafka topic

You can skip this if the topic is already present. Before we can produce a message, we need to have a topic. This function also checks if the topic is already present in case you run this multiple times.

// create the kafka topic where we are going to produce the data
const createTopic = async () => {
  try {
    const topicExists = (await kafka.admin().listTopics()).includes(TOPIC);
    if (!topicExists) {
      await kafka.admin().createTopics({
        topics: [
          {
            topic: TOPIC,
            numPartitions: 1,
            replicationFactor: 1,
          },
        ],
      });
    }
  } catch (error) {
    console.log(error);
  }
};

Now we create our producer and consumer functions which publish an example message and consume it again.

const produce = async () => {
  await createTopic();
  try {
    const registryId = await registerSchema();
    // push example message
    if (registryId) {
      const message: MyMessage = { id: "1", value: 1 };
      await produceToKafka(registryId, message);
      console.log(`Produced message to Kafka: ${JSON.stringify(message)}`);
    }
  } catch (error) {
    console.log(`There was an error producing the message: ${error}`);
  }
};

async function consume() {
  await consumer.connect();

  await consumer.subscribe({
    topic: TOPIC,
    fromBeginning: true,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      if (message.value) {
        const value: MyMessage = await registry.decode(message.value);
        console.log(value);
      }
    },
  });
}

And finally we execute both functions one after another.

produce()
  .then(() => consume())

The console should print something like:

Produced message to Kafka: {"id":"1","value":1}
Consumed message from Kafka: Example { id: '1', value: 1 }

Demo repository with this code

I created a repository to demo this example. There is a docker-compose file which takes care of setting up a Kafka Broker and a Schema Registry.

Throttle a series of fetch requests in JavaScript

Eduard Gert — Mon, 01 Mar 2021 09:19:23 +0000

Let's say you need to make API requests to process a huge array of data. With JavaScript's asynchronous nature it is easy to make a lot of requests in parallel.

import fetch from "node-fetch";

const data = [{ id: 1 }, { id: 2 }, [+1000 more objects]];

const fetchFromApi = (id) => {
  const url = `https://example.com/api/my-resource/${id}`;

  const response = fetch(url)
    .then((x) => x.json())
    .catch((error) => console.log(error));
  return response;
};

for (const i of data) {
  fetchFromApi(i.id).then((result) => // do something with result);
}

HTTP code 429: Too many requests

However, most API providers don't like if you flood them with too many requests at the same time.
What you usually would get in return is a HTTP error code 429. If you check the documentation there might be a limitation of let's say maximum 5 requests per second.
But even if it is an internal API that isn't that restricted you might want to reduce the amount of parallel requests.

Wait for the response before making another request?

What you could do is introducing a blocking structure to wait for the response of the previous call, before making another one using JavaScripts async/await syntax.

import fetch from "node-fetch";

const data = [{ id: 1 }, { id: 2 }, [+1000 more objects]];

const fetchFromApi = async (id) => {
  const url = `https://example.com/api/my-resource/${id}`;

  const response = fetch(url)
    .then((x) => x.json())
    .catch((error) => console.log(error));
  return response;
};

for (const i of data) {
  const response = await fetchFromApi(i.id);
  // do something with result
}

While this would take longer to run, it would not solve the issue. The API might respond very quickly and you would still reach the limit of 5 requests per second.
On the other hand if the API responds slowly, you wouldn't benefit of parallelism at all, which would make the whole operation take longer than needed.

Semaphore to the rescue

Using a throttling mechanism would be the more elegant way to deal with this issue. In computer science there's the concept of a semaphore which describes a way to control access to a common resource by multiple processes.
There is a library which implements that and allows you to limit the maximum parallel requests. The code would look something like this:

import fetch from "node-fetch";
import {RateLimit} from "async-sema";

// configure a limit of maximum 5 requests / second
const limit = RateLimit(5);

const data = [{ id: 1 }, { id: 2 }, [+1000 more objects]];

const fetchFromApi = (id) => {
  const url = `https://example.com/api/my-resource/${id}`;

  // use the configured throttle here
  const response = fetch(url)
    .then((x) => x.json())
    .catch((error) => console.log(error));
  return response;
};

for (const i of data) {
  // checks if limit is reached
  await limit()
  fetchFromApi(i.id).then((result) => console.log(result));
}

DEV Community: Eduard Gert