How to Load Test Your Kafka Producers and Consumers using k6
Tags: performance, javascript, kafka, testing
Published: 2021-09-22 01:34:25

Recently, k6 started supporting [k6 extensions](https://github.com/grafana/xk6) to extend k6 capabilities for other cases required by the community. The community has already built [plenty of extensions](https://k6.io/docs/ecosystem/). k6 extensions are written in Go, and many of them are reusing existing Go libraries. 

This makes k6 to be a versatile tool to test different protocols and adapt to multiple cases. This post is the third part of my series of articles testing various systems using k6:

- [Load Testing SQL Databases](https://dev.to/k6/load-testing-sql-databases-with-k6-30ci)
- [Benchmarking Redis](https://dev.to/k6/benchmarking-redis-with-k6-33kf)

Let's look in this post how we test the popular Kafka project. [Apache Kafka](https://kafka.apache.org/intro) is a powerful event streaming platform that provides the following features:

- Write and read streams of events
- Store streams of events for as long as you want to
- Process streams of events in parallel retrospectively

It works by having client applications write events to the Kafka server. We term this type of application as **Producers**. Client applications which read and process events from the Kafka server are called **Consumers**.

Kafka itself is capable of handling hundreds to millions of events per second seamlessly on simple setup.  But what if you wanted to test and observe how your Kafka service behaves before going live?

The [xk6-kafka extension](https://github.com/mostafa/xk6-kafka) provides some convenient functions for interacting with Kafka Producers and Consumers. It serves as a producer that can send a high volume of messages per second, allowing you to monitor the system under test (SUT) and test how the application will keep up with the load.

## xk6-kafka

At the time of this writing, the xk6-kafka extension provides the following APIs:

| Function                   | Description                   |
| -------------------------- | ----------------------- |
| consume(reader, limit) | Consume messages from the Kafka server.  |
| createTopic(address, topic) | Create a new topic.   |
| listTopics(address) | Return a unique set of topics.   |
| produce(writer, messages) | Produce messages to the Kafka server. |
| reader(brokers, topic) | Instantiate a new Reader instance. |
| writer(brokers, topic) | Instantiate a new Writer instance. |


Some of the APIs mentioned above do accept additional optional parameters meant for authentication and message compression. Refer to [more examples](#more-examples) for additional information.

### Building k6 with the kafka extension

By default, k6 does not support testing Kafka. Building k6 with the [xk6-kafka extension](https://github.com/mostafa/xk6-kafka) creates a k6 version with the capabilities to test Kafka producers and consumers.

Make sure you have the following installed and ready before you continue:
- Go (>=1.7)
- Git

Next, continue the installation by running the following command in your terminal to install the xk6 module:

```bash
go install go.k6.io/xk6/cmd/xk6@latest
```

Once the command finishes successfully, you can start making your own custom k6 binary for Kafka as follows:

```bash
xk6 build --with github.com/mostafa/xk6-kafka@latest
```

It will take some time for the process to create a new k6 binary in your working directory.

> Troubleshooting: If you experienced issues installing xk6 extension, kindly head over to the [Github repository](https://github.com/grafana/xk6/releases/) to download the pre-compiled binaries instead.


## Running Kafka

The recommended approach is to use docker as manual installation is quite complicated and prone to errors. You can pull the [following image by lensesio](https://hub.docker.com/r/lensesio/fast-data-dev/tags?page=1&ordering=last_updated) from DockerHub. It contains the complete Kafka setup for development.

```bash
docker pull lensesio/fast-data-dev:latest
```

After that, run the following to start the docker in detached mode:

```bash
sudo docker run -d --rm --name lenseio -p 2181:2181 -p 3030:3030 \
       -p 8081-8083:8081-8083 -p 9581-9585:9581-9585 -p 9092:9092  \
       -e ADV_HOST=127.0.0.1 lensesio/fast-data-dev

sudo docker logs -f -t lenseio
```

Visit [`http://localhost:3030`](http://localhost:3030) to get into the fast-data-dev environment.

## k6 test

### Import

Now, let’s create a new JavaScript file called test_script.js in the same directory as your k6 binary. Then, add the following import statement at the top of the file:

```javascript
import { check } from "k6";
import { writer, produce, reader, consume, createTopic } from "k6/x/kafka";
```

### Initialization

Continue by appending the following initialization code:

```javascript
const bootstrapServers = ["localhost:9092"];
const kafkaTopic = "xk6_kafka_json_topic";

const producer = writer(bootstrapServers, kafkaTopic);
const consumer = reader(bootstrapServers, kafkaTopic);
```

The code will initialize both the writer and reader instances based on the configuration specified. If you are using a different IP/host address and port for your Kafka server, kindly modify it accordingly.

Next, call the createTopic function to create a new topic. Rest assured that this function will do nothing if the topic already exists.


```javascript
createTopic(bootstrapServers[0], kafkaTopic);
```

Let’s create a function which generates a random integer as a unique identifier for each message later on. Please be noted that this is optional and not a mandatory requirement to do load testing.


```javascript
function getRandomInt(max=1000) {
  return Math.floor((Math.random() * max) + 1);
}
```

### Default function

As for the default function, define it as follows:

```javascript
export default function () {
	let messages = [{
      key: JSON.stringify({
          correlationId: "test-id-sql-" + getRandomInt(),
      }),
      value: JSON.stringify({
          title: "Load Testing SQL Databases with k6",
          url: "https://k6.io/blog/load-testing-sql-databases-with-k6/",
          locale: "en"
      }),
    },
    {
      key: JSON.stringify({
          correlationId: "test-id-redis-" + getRandomInt(),
      }),
      value: JSON.stringify({
          title: "Benchmarking Redis with k6",
          url: "https://k6.io/blog/benchmarking-redis-with-k6/",
          locale: "en"
      }),
  }];

	let error = produce(producer, messages);
	check(error, {
    	  "is sent": (err) => err == undefined,
	});
}
```

The code block above works as follows:
- Initialize a list of messages
- Call produce function to publish the messages
- Check if messages are successfully sent

### Teardown

Once you are done with it, create a teardown function and close the connections:

```javascript
export function teardown(data) {
	producer.close();
	consumer.close();
}
```

### Run the test

Save the file and run the following command on your terminal:

```bash
./k6 run --vus 50 --duration 5s test_script.js
```

You should see the following output:

```bash
running (05.0s), 00/50 VUs, 15136 complete and 0 interrupted iterations
default ✓ [======================================] 50 VUs  5s

 	✓ is sent

 	█ teardown

 	checks.........................: 100.00% ✓ 15136 	✗ 0
 	data_received..................: 0 B 	0 B/s
 	data_sent......................: 0 B 	0 B/s
 	iteration_duration.............: avg=16.49ms min=31.9µs med=13.52ms max=1.14s p(90)=28.55ms p(95)=36.46ms
 	iterations.....................: 15136   3017.4609/s
 	kafka.writer.dial.count........: 151 	30.102841/s
 	kafka.writer.error.count.......: 0   	0/s
 	kafka.writer.message.bytes.....: 5.2 MB  1.0 MB/s
 	kafka.writer.message.count.....: 30272   6034.9218/s
 	kafka.writer.rebalance.count...: 0   	0/s
 	kafka.writer.write.count.......: 30272   6034.9218/s
 	vus............................: 5   	min=5   	max=50
 	vus_max........................: 50  	min=50  	max=50
```

### Scale the load

You can easily scale the load by increasing the number of vus. For example, the following command uses 500 vus to load test for a minute:

```bash
./k6 run --vus 500 --duration 1m test_script.js
```

> If you are new to k6, check out how to configure the [load options](https://k6.io/docs/getting-started/running-k6/#using-options) in the script or run a [stress test](https://k6.io/docs/test-types/stress-testing/) with k6.

### Extend the test

The script above is all about producing messages to your Kafka server. In fact, you can easily modify the code into a test which produces and consumes messages.

Simply add the following code below the for loop code:

```bash
let result = consume(consumer, 10);
check(result, {
    "10 messages returned": (msgs) => msgs.length == 10,
});
```

The code will read 10 messages each time. Simply modify the value to something higher if you wish to consume more messages.

The output is as follows when you run it with the same command:

```bash
running (05.0s), 00/50 VUs, 9778 complete and 0 interrupted iterations
default ✓ [======================================] 50 VUs  5s

 	✓ is sent
 	✓ 10 messages returned

 	█ teardown

 	checks.........................: 100.00% ✓ 19556    	✗ 0
 	data_received..................: 0 B 	0 B/s
 	data_sent......................: 0 B 	0 B/s
 	iteration_duration.............: avg=25.53ms min=41.4µs med=18ms max=1.41s p(90)=37.73ms p(95)=52.37ms
 	iterations.....................: 9778	1946.80798/s
 	kafka.reader.dial.count........: 50  	9.955042/s
 	kafka.reader.error.count.......: 0   	0/s
 	kafka.reader.fetches.count.....: 101 	20.109184/s
 	kafka.reader.message.bytes.....: 15 MB   2.9 MB/s
 	kafka.reader.message.count.....: 97830   19478.034846/s
 	kafka.reader.rebalance.count...: 0   	0/s
 	kafka.reader.timeouts.count....: 46  	9.158638/s
 	kafka.writer.dial.count........: 152 	30.263327/s
 	kafka.writer.error.count.......: 0   	0/s
 	kafka.writer.message.bytes.....: 3.4 MB  669 kB/s
 	kafka.writer.message.count.....: 19556   3893.615961/s
 	kafka.writer.rebalance.count...: 0   	0/s
 	kafka.writer.write.count.......: 19556   3893.615961/s
 	vus............................: 50  	min=50     	max=50
 	vus_max........................: 50  	min=50     	max=50
```

## Kafka metrics in k6

By default, k6 has its own [built-in metrics](https://k6.io/docs/using-k6/metrics/) which are collected automatically. Apart from that, you can create your [own custom metrics](https://k6.io/docs/using-k6/metrics/#custom-metrics). Custom metrics can be categorized into the following types:

- [Counter](https://k6.io/docs/javascript-api/k6-metrics/counter): A metric that cumulatively sums added values.
- [Gauge](https://k6.io/docs/javascript-api/k6-metrics/gauge): A metric that stores the min, max and last values added to it. 
- [Rate](https://k6.io/docs/javascript-api/k6-metrics/rate): A metric that tracks the percentage of added values that are non-zero.
- [Trend](https://k6.io/docs/javascript-api/k6-metrics/trend): A metric that allows for calculating statistics on the added values (min, max, average and percentiles). 

Besides k6, k6 extensions can collect metrics and report them as part of the [k6 results output](https://k6.io/docs/getting-started/results-output/). In this case, xk6-kafka collects individual stats for both reader and writer. 

### Reader

Let’s have a look at the metrics meant for the [reader](https://github.com/mostafa/xk6-kafka/blob/697dde73c0979eab0187dd81addbfd11ca8da448/consumer.go#L138).


| Metrics                   | Type | Description                   |
| -------------------------- | ----- | ----------------------- |
| kafka.reader.dial.count  | Counter | Total number of times the reader tries to connect to Kafka. |
| kafka.reader.error.count | Counter | Total number of errors occurred when reading from Kafka.  |
| kafka.reader.fetches.count | Counter | Total number of times the reader fetches batches of messages from Kafka.  |
| kafka.reader.message.bytes | Counter |  Total bytes consumed. |
| kafka.reader.message.count | Counter |  Total number of messages consumed. |
| kafka.reader.rebalance.count | Counter | Total number of rebalances of a topic in a consumer group (_deprecated_) .  |
| kafka.reader.timeouts.count | Counter |  Total number of timeouts occurred when reading from Kafka |

### Writer

As for the [writer](https://github.com/mostafa/xk6-kafka/blob/697dde73c0979eab0187dd81addbfd11ca8da448/producer.go#L135), the metrics are as follows:

| Metrics                   | Type | Description                   |
| -------------------------- | ----- | ----------------------- |
| kafka.writer.dial.count  | Counter | Total number of times the writer tries to connect to Kafka. |
| kafka.writer.error.count  | Counter | Total number of errors occurred when writing to Kafka. |
| kafka.writer.message.bytes | Counter | Total bytes produced. |
| kafka.writer.message.count | Counter | Total number of messages produced. |
| kafka.writer.rebalance.count | Counter | Total number of rebalances of a topic (_deprecated_). |
| kafka.writer.write.count | Counter |  Total number of times the writer writes batches of messages to Kafka. |


There are more available kafka metrics, as you can find them [here](https://github.com/mostafa/xk6-kafka/blob/master/stats.go). However, the extension does not collect all metrics yet. You can follow this [GitHub issue](https://github.com/mostafa/xk6-kafka/issues/23) to track the progress of their additions.


## More examples 

Moreover, the xk6-kafka repository provides a [few test scripts](https://github.com/mostafa/xk6-kafka/tree/master/scripts) that work out-of-the-box for new users. At the time of this writing, it comes with the following tests:

- [test_avro.js](https://github.com/mostafa/xk6-kafka/blob/master/scripts/test_avro.js) : Tests Kafka with 200 Avro messages per iteration.
- [test_avro_with_schema_registry.js](https://github.com/mostafa/xk6-kafka/blob/master/scripts/test_avro_with_schema_registry.js) : Tests Kafka with 200 Avro messages per iteration using schema registry
- [test_json.js](https://github.com/mostafa/xk6-kafka/blob/master/scripts/test_json.js) : Tests Kafka with 200 JSON messages per iteration.
- [test_json_with_snappy_compression.js](https://github.com/mostafa/xk6-kafka/blob/master/scripts/test_json_with_snappy_compression.js) : Tests Kafka with 200 JSON messages per iteration using snappy compression
- [test_sasl_auth.js](https://github.com/mostafa/xk6-kafka/blob/master/scripts/test_sasl_auth.js) : Tests Kafka with 200 JSON messages per iteration and based on SASL authentication.
- [test_topics.js](https://github.com/mostafa/xk6-kafka/blob/master/scripts/test_topics.js) : List topics on all Kafka partitions and create random topics.

Feel free to experiment with them and modify the code accordingly based on your own use cases. If you bump into issues, report them [on GitHub](https://github.com/mostafa/xk6-kafka/issues).


## Conclusion

In conclusion, load testing Apache Kafka is now a lot easier with k6. k6 provides the foundation to create and scale your load tests, and the xk6-kafka extension brings a convenient API to interact with a Kafka server.

If you wish to find out more on other available k6 extensions, simply head over to the [bundle builder page](https://k6.io/docs/ecosystem/bundle-builder/). The page also allows you to generate the corresponding command for building your own custom k6 binary.

If you have any questions or are interested in building an extension, join the k6 community on Slack.