mostafamoradian/xk6-kafka repository overview

⁠xk6-kafka

The xk6-kafka project is a k6 extension⁠ that enables k6 users to load test Apache Kafka using a producer and possibly a consumer for debugging.

The real purpose of this extension is to test the system you meticulously designed to use Apache Kafka. So, you can test your consumers, hence your system, by auto-generating messages and sending them to your system via Apache Kafka.

You can send many messages with each connection to Kafka. These messages are arrays of objects containing a key and a value in various serialization formats, passed via configuration objects. Various serialization formats are supported, including strings, JSON, binary, Avro, and JSON Schema. Avro and JSON Schema can either be fetched from Schema Registry or hard-code directly in the script. SASL PLAIN/SCRAM authentication and message compression are also supported.

For debugging and testing purposes, a consumer is available to make sure you send the correct data to Kafka.

If you want to learn more about the extension, read the article⁠ (outdated) explaining how to load test your Kafka producers and consumers using k6 on the k6 blog. You can also watch this recording⁠ of the k6 Office Hours about this extension.

⁠Supported Features

• Produce/consume messages as String⁠, JSON⁠, ByteArray⁠, Avro⁠ and JSON Schema⁠ formats
• Support for user-provided Avro⁠ and JSON Schema key and value schemas in the script
• Authentication with SASL PLAIN, SCRAM and SSL⁠
• Create, list and delete topics⁠
• Support for loading Avro schemas from Schema Registry⁠
• Support for byte array⁠ for binary data (from binary protocols)
• Support consumption from all partitions with a group ID⁠
• Support Kafka message compression: Gzip, Snappy⁠, Lz4 & Zstd
• Support for sending messages with no key⁠
• Support for k6 thresholds⁠ on custom Kafka metrics
• Support for headers⁠ on produced and consumed messages
• Lots of exported metrics, as shown in the result output of the k6 test script⁠

⁠Download Binaries

⁠The Official Docker Image

The official Docker image⁠ is available on Docker Hub. Before running your script, make the script available to the container by mounting a volume (a directory) or passing it via stdin.

docker run --rm -i mostafamoradian/xk6-kafka:latest run - <scripts/test_json.js

⁠The Official Binaries

The binaries are generated by the build process and published on the releases page⁠. Currently, binaries for the GNU/Linux, macOS, and Windows on amd64 (x86_64) machines are available.

⁠Build from Source

You can build the k6 binary on various platforms, each with its requirements. The following shows how to build k6 binary with this extension on GNU/Linux distributions.

⁠Prerequisites

You must have the latest Go version installed to build the k6 binary. The latest version should match k6⁠ and xk6⁠. I recommend gvm⁠ because it eases version management.

• gvm⁠ for easier installation and management of Go versions on your machine
• Git⁠ for cloning the project
• xk6⁠ for building k6 binary with extensions

⁠Install and build the latest tagged version

Feel free to skip the first two steps if you already have Go installed.

1. Install gvm by following its installation guide⁠.

2. Install the latest version of Go using gvm. You need Go 1.4 installed for bootstrapping into higher Go versions, as explained here⁠.

3. Install xk6:

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

   Copy

4. Build the binary:

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

   Copy

⁠Build for development

If you want to add a feature or make a fix, clone the project and build it using the following commands. The xk6 will force the build to use the local clone instead of fetching the latest version from the repository. This process enables you to update the code and test it locally.

git clone [email protected]:mostafa/xk6-kafka.git && cd xk6-kafka
xk6 build --with github.com/mostafa/xk6-kafka@latest=.

⁠Example scripts

There are many examples in the script⁠ directory that show how to use various features of the extension.

⁠How to Test You Kafka Setup

You can start testing your setup immediately, but it takes some time to develop the script, so it would be better to test your script against a development environment and then start testing your environment.

⁠Development environment

I recommend the fast-data-dev⁠ Docker image by Lenses.io, a Kafka setup for development that includes Kafka, Zookeeper, Schema Registry, Kafka-Connect, Landoop Tools, 20+ connectors. It is relatively easy to set up if you have Docker installed. Just monitor Docker logs to have a working setup before attempting to test because the initial setup, leader election, and test data ingestion take time.

1. Run the Kafka environment and expose the ports:

   sudo docker run \
       --detach --rm \
       --name lensesio \
       -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 \
       -e RUN_TESTS=0 \
       lensesio/fast-data-dev:latest
   

   Copy

2. After running the command, visit localhost:3030⁠ to get into the fast-data-dev environment.

3. You can run the command to see the container logs:

   sudo docker logs -f -t lensesio
   

   Copy

⁠The xk6-kafka API

All the exported functions are available by importing the module object from k6/x/kafka. The exported objects, constants and other data structures are available in the index.d.ts⁠ file, and they always reflect the latest changes on the main branch. You can access the generated documentation at api-docs/docs/README.md⁠.

⁠k6 Test Scripts

The example scripts are available as test_<format/feature>.js with more code and commented sections in the scripts⁠ directory. Since this project extends the functionality of k6, it has four stages in the test life cycle⁠.

1. To use the extension, you need to import it in your script, like any other JS module:

   // Either import the module object
   import * as kafka from "k6/x/kafka";
   
   // Or individual classes and constants
   import {
     Writer,
     Reader,
     Connection,
     SchemaRegistry,
     SCHEMA_TYPE_STRING,
   } from "k6/x/kafka";
   

   Copy

2. You need to instantiate the classes in the init context. All the k6 options⁠ are also configured here:

   // Creates a new Writer object to produce messages to Kafka
   const writer = new Writer({
     // WriterConfig object
     brokers: ["localhost:9092"],
     topic: "my-topic",
   });
   
   const reader = new Reader({
     // ReaderConfig object
     brokers: ["localhost:9092"],
     topic: "my-topic",
   });
   
   const connection = new Connection({
     // ConnectionConfig object
     address: "localhost:9092",
   });
   
   const schemaRegistry = new SchemaRegistry();
   // Can accept a SchemaRegistryConfig object
   
   if (__VU == 0) {
     // Create a topic on initialization (before producing messages)
     connection.createTopic({
       // TopicConfig object
       topic: "my-topic",
     });
   }
   

   Copy

3. In the VU code, you can produce messages to Kafka or consume messages from it:

   export default function () {
     // Fetch the list of all topics
     const topics = connection.listTopics();
     console.log(topics); // list of topics
   
     // Produces message to Kafka
     writer.produce({
       // ProduceConfig object
       messages: [
         // Message object(s)
         {
           key: schemaRegistry.serialize({
             data: "my-key",
             schemaType: SCHEMA_TYPE_STRING,
           }),
           value: schemaRegistry.serialize({
             data: "my-value",
             schemaType: SCHEMA_TYPE_STRING,
           }),
         },
       ],
     });
   
     // Consume messages from Kafka
     let messages = reader.consume({
       // ConsumeConfig object
       limit: 10,
     });
   
     // your messages
     console.log(messages);
   
     // You can use checks to verify the contents,
     // length and other properties of the message(s)
   
     // To serialize the data back into a string, you should use
     // the deserialize method of the Schema Registry client. You
     // can use it inside a check, as shown in the example scripts.
     let deserializedValue = schemaRegistry.deserialize({
       data: messages[0].value,
       schemaType: SCHEMA_TYPE_STRING,
     });
   }
   

   Copy

4. In the teardown function, close all the connections and possibly delete the topic:

   export function teardown(data) {
     // Delete the topic
     connection.deleteTopic("my-topic");
   
     // Close all connections
     writer.close();
     reader.close();
     connection.close();
   }
   

   Copy

5. You can now run k6 with the extension using the following command:

   ./k6 run --vus 50 --duration 60s scripts/test_json.js
   

   Copy

6. And here's the test result output:

   
           /\      |‾‾| /‾‾/   /‾‾/
      /\  /  \     |  |/  /   /  /
     /  \/    \    |     (   /   ‾‾\
    /          \   |  |\  \ |  (‾)  |
   / __________ \  |__| \__\ \_____/ .io
   
   execution: local
       script: scripts/test_json.js
       output: -
   
   scenarios: (100.00%) 1 scenario, 50 max VUs, 1m30s max duration (incl. graceful stop):
           * default: 50 looping VUs for 1m0s (gracefulStop: 30s)
   
   
   running (1m04.4s), 00/50 VUs, 20170 complete and 0 interrupted iterations
   default ✓ [======================================] 50 VUs  1m0s
   
       ✓ 10 messages are received
       ✓ Topic equals to xk6_kafka_json_topic
       ✓ Key contains key/value and is JSON
       ✓ Value contains key/value and is JSON
       ✓ Header equals {'mykey': 'myvalue'}
       ✓ Time is past
       ✓ Partition is zero
       ✓ Offset is gte zero
       ✓ High watermark is gte zero
   
       █ teardown
   
       checks.........................: 100.00% ✓ 181530       ✗ 0
       data_received..................: 0 B     0 B/s
       data_sent......................: 0 B     0 B/s
       iteration_duration.............: avg=153.45ms min=6.01ms med=26.8ms  max=8.14s   p(90)=156.3ms p(95)=206.4ms
       iterations.....................: 20170   313.068545/s
       kafka_reader_dial_count........: 50      0.776075/s
       kafka_reader_dial_seconds......: avg=171.22µs min=0s     med=0s      max=1.09s   p(90)=0s      p(95)=0s
     ✓ kafka_reader_error_count.......: 0       0/s
       kafka_reader_fetch_bytes_max...: 1000000 min=1000000    max=1000000
       kafka_reader_fetch_bytes_min...: 1       min=1          max=1
       kafka_reader_fetch_wait_max....: 200ms   min=200ms      max=200ms
       kafka_reader_fetch_bytes.......: 58 MB   897 kB/s
       kafka_reader_fetch_size........: 147167  2284.25179/s
       kafka_reader_fetches_count.....: 107     1.6608/s
       kafka_reader_lag...............: 1519055 min=0          max=2436190
       kafka_reader_message_bytes.....: 40 MB   615 kB/s
       kafka_reader_message_count.....: 201749  3131.446006/s
       kafka_reader_offset............: 4130    min=11         max=5130
       kafka_reader_queue_capacity....: 1       min=1          max=1
       kafka_reader_queue_length......: 1       min=0          max=1
       kafka_reader_read_seconds......: avg=96.5ms   min=0s     med=0s      max=59.37s  p(90)=0s      p(95)=0s
       kafka_reader_rebalance_count...: 0       0/s
       kafka_reader_timeouts_count....: 57      0.884725/s
       kafka_reader_wait_seconds......: avg=102.71µs min=0s     med=0s      max=85.71ms p(90)=0s      p(95)=0s
       kafka_writer_acks_required.....: 0       min=0          max=0
       kafka_writer_async.............: 0.00%   ✓ 0            ✗ 2017000
       kafka_writer_attempts_max......: 0       min=0          max=0
       kafka_writer_batch_bytes.......: 441 MB  6.8 MB/s
       kafka_writer_batch_max.........: 1       min=1          max=1
       kafka_writer_batch_size........: 2017000 31306.854525/s
       kafka_writer_batch_timeout.....: 0s      min=0s         max=0s
     ✓ kafka_writer_error_count.......: 0       0/s
       kafka_writer_message_bytes.....: 883 MB  14 MB/s
       kafka_writer_message_count.....: 4034000 62613.709051/s
       kafka_writer_read_timeout......: 0s      min=0s         max=0s
       kafka_writer_retries_count.....: 0       0/s
       kafka_writer_wait_seconds......: avg=0s       min=0s     med=0s      max=0s      p(90)=0s      p(95)=0s
       kafka_writer_write_count.......: 4034000 62613.709051/s
       kafka_writer_write_seconds.....: avg=523.21µs min=4.84µs med=14.48µs max=4.05s   p(90)=33.85µs p(95)=42.68µs
       kafka_writer_write_timeout.....: 0s      min=0s         max=0s
       vus............................: 7       min=7          max=50
       vus_max........................: 50      min=50         max=50
   

   Copy

⁠Emitted Metrics

⁠FAQ

1. Why do I receive Error writing messages?

   There are a few reasons why this might happen. The most prominent one is that the topic might not exist, which causes the producer to fail to send messages to a non-existent topic. You can use Connection.createTopic method to create the topic in Kafka, as shown in scripts/test_topics.js. You can also set the autoCreateTopic on the WriterConfig. You can also create a topic using the kafka-topics command:

   $ docker exec -it lensesio bash
   (inside container)$ kafka-topics --create --topic xk6_kafka_avro_topic --bootstrap-server localhost:9092
   (inside container)$ kafka-topics --create --topic xk6_kafka_json_topic --bootstrap-server localhost:9092
   

   Copy

2. Why does the reader.consume keep hanging?

   If the reader.consume keeps hanging, it might be because the topic doesn't exist or is empty.

3. I want to test SASL authentication. How should I do that?

   If you want to test SASL authentication, look at this commit message⁠, in which I describe how to run a test environment to test SASL authentication.

4. Why doesn't the consumer group consume messages from the topic?

   As explained in issue #37⁠, multiple inits by k6 cause multiple consumer group instances to be created in the init context, which sometimes causes the random partitions to be selected by each instance. This, in turn, causes confusion when consuming messages from different partitions. This can be solved by using a UUID when naming the consumer group, thereby guaranteeing that the consumer group object was assigned to all partitions in a topic.

5. Why do I receive a MessageTooLargeError when I produce messages bigger than 1 MB?

   Kafka has a maximum message size⁠ of 1 MB by default, which is set by message.max.bytes, and this limit is also applied to the Writer object.

   There are two ways to produce larger messages: 1) Change the default value of your Kafka instance to a larger number. 2) Use compression.

   Remember that the Writer object will reject messages larger than the default Kafka message size limit (1 MB). Hence you need to set batchBytes to a larger value, for example, 1024 * 1024 * 2 (2 MB). The batchBytes refers to the raw uncompressed size of all the keys and values (data) in your array of messages you pass to the Writer object. You can calculate the raw data size of your messages using this example script⁠.

6. Can I consume messages from a consumer group in a topic with multiple partitions?

   Yes, you can. Just pass the groupID to your Reader object. You must not specify the partition anymore. Visit this documentation article⁠ to learn more about Kafka consumer groups.

   Remember that you must set sessionTimeout on your Reader object if the consume function terminates abruptly, thus failing to consume messages.

7. Why does the Reader.consume produces an unable to read message error?

   For performance testing reasons, the maxWait of the Reader is set to 200ms. If you keep receiving this error, consider increasing it to a larger value.

8. How can I consume from multiple partitions on a single topic?

   You can configure your reader to consume from a (list of) topic(s) and its partitions using a consumer group. This can be achieve by setting groupTopics, groupID and a few other options for timeouts, intervals and lags. Have a look at the test_consumer_group.js⁠ example script.

⁠Contributions, Issues and