k6 kafka

k6_kafka.js

/*

This is a k6 test script that imports the xk6-kafka and
tests Kafka with a 200 JSON messages per iteration.

*/

import { check } from "k6";
// import * as kafka from "k6/x/kafka";
import {
  Writer,
  Reader,
  Connection,
  SchemaRegistry,
  CODEC_SNAPPY,
  SCHEMA_TYPE_JSON,
} from "k6/x/kafka"; // import kafka extension

// Prints module-level constants
// console.log(kafka);

const brokers = ["localhost:9092"];
const topic = "user.new";

const writer = new Writer({
  brokers: brokers,
  topic: topic,
  autoCreateTopic: true,
  compression: CODEC_SNAPPY,
});


const connection = new Connection({
  address: brokers[0],
});


const schemaRegistry = new SchemaRegistry();

if (__VU == 0) {
  connection.createTopic({
    topic: topic,
    configEntries: [
      {
        configName: "compression.type",
        configValue: CODEC_SNAPPY,
      },
    ],
  });
}

export const options = {
  thresholds: {
    // Base thresholds to see if the writer or reader is working
    kafka_writer_error_count: ["count == 0"],
  },
};

export default function () {
  for (let index = 0; index < 10000; index++) {
    let messages = [
      {
        // The data type of the value is JSON
        value: schemaRegistry.serialize({
          data: {
			  "header": {
				"version": "0.1",
				"sessionId": "wertyuiopsdfghjklvbnm"
			  },
			  "body": {
				"firstname": "123",
				 "lastname": "45"
			  }
			},
          schemaType: SCHEMA_TYPE_JSON,
        }),
        offset: index,
        partition: 0,
        time: new Date(), // Will be converted to timestamp automatically
      },
    ];

    writer.produce({ messages: messages });
  }


}

export function teardown(data) {
  if (__VU == 0) {
    // Delete the topic
    connection.deleteTopic(topic);
  }
  writer.close();
  connection.close();
}