Kafka Load Test

Welcome

Some of our users have inquired as to whether we can do a load test of their Kafka cluster using Gatling. You’ve asked, we’ve delivered: here’s a step-by-step guide!

Apache Kafka is an open-source, real-time data streaming platform. the platform can collect, store, and distribute data streams (events) across large-scale applications and systems.

More specifically, Kafka works as a distributed messaging system, sending data as messages between producers (message creators) and consumers (message readers). Messages are stored in partitions and distributed across multiple nodes in a Kafka server cluster.

With the help of our active community, we’ve provided a plugin to perform this type of load test: https://github.com/Tinkoff/gatling-kafka-plugin.

The project is configured to work with sbt. It can also be ported to other build tools like Maven or Gradle.

Overloading a Kafka broker can lead to latency, message loss, and even crashes. Load testing helps identify these bottlenecks and determine the maximum load a Kafka broker can handle before performance degradation. By pinpointing these issues at an early stage, you can optimize your Kafka broker to ensure it meet your performance and scalability requirements.

Prerequisites

This guide will help you set up a load test on a Kafka broker. Our prerequisites are:  

• Java 17
• Maven or Gradle
• A Kafka broker (up and running)
  • How to install a Kafka cluster with Docker 
• An account on gatling.cloud.io

To simplify the installation, we recommend you use the public demo project for Gatling in Java with one of the following: Maven, or Gradle.

Add the dependencies

The first step is to add the dependencies to your project.

For Maven

Add a new repository to your .pom file

<repositories>
 <repository>
   <id>confluent</id>
   <url>https://packages.confluent.io/maven/</url>
 </repository>
</repositories>

 Add this dependency to your .pom file

<dependency>
 <groupId>ru.tinkoff</groupId>
 <artifactId>gatling-kafka-plugin_2.13</artifactId>
 <version>0.11.0</version>
 <scope>test</scope>
</dependency>

For Gradle

Add a new repository to your build.gradle file 

repositories {
  mavenCentral()
    maven {
        url "https://packages.confluent.io/maven/"
    }
}

Add this dependency to your build.gradle file 

dependencies {
    gatling "ru.tinkoff:gatling-kafka-plugin_2.13:0.11.0"
}

Add the imports

Create a new KafkaSimulation class. Next, you’ll need to add the imports to your simulation class so you can use the Kafka plugin.

import ru.tinkoff.gatling.kafka.javaapi.protocol.*;
import static io.gatling.javaapi.core.CoreDsl.*;
import static ru.tinkoff.gatling.kafka.javaapi.KafkaDsl.*;

Define your cluster configuration

Define the configuration for your Kafka cluster (at least the cluster URL and topic name(s)).

private final KafkaProtocolBuilder kafkaProtocol = kafka()
       .topic("test")
       .properties(
               Map.of(
                       ProducerConfig.ACKS_CONFIG, "1",
                       ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
                       ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer",
                       ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG , "org.apache.kafka.common.serialization.StringSerializer")
       );

If you use authentication with your Kafka cluster (e.g. SASL), you need to define it here. In a complex configuration like above, you can take advantage of Gatling’s flexibility to create environment variables. It will allow you to organize, maintain, and change the Kafka cluster configuration more easily. 

For example:

public static final String IP_SERVER = System.getProperty("IP_SERVER", "");
public static final String URL_REGISTRY = System.getProperty("URL_REGISTRY", "");
public static final String USER_AUTH = System.getProperty("USER_AUTH", "");
private final KafkaProtocolBuilder kafkaProtocol = kafka()
       .topic("test")
       .properties(
               Map.of(
                       ProducerConfig.ACKS_CONFIG, "1",
                       ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, IP_SERVER,
                       ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "2097152",
                       "security.protocol", "SASL_SSL",
                       "sasl.mechanism", "PLAIN",
                       "client.dns.lookup", "use_all_dns_ips",
                       "sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$USERNAME_BROKER\" password=\"$PASSWORD_BROKER\";",
                       "schema.registry.url",URL_REGISTRY,
                       "basic.auth.credentials.source","USER_INFO",
                       "basic.auth.user.info", USER_AUTH
       )
       );

Test scenario

Let’s take a simple example where we push a text message with a header to our cluster. 

private final Headers headers = new RecordHeaders(new Header[]{new RecordHeader("test-header", "value".getBytes())});
private final ScenarioBuilder kafkaProducer = scenario("Kafka Producer")
   .exec(kafka("Simple Message")
           .send("key","value", headers)
   );

We can then design other processes to ensure that system components have properly consumed the message. Even though Kafka is designed to work in an asynchronous model, you can use the Kafka plugin to create complex scenarios, especially for request-reply. 

You can also build messages in different formats. For example, you can use Avro libraries to serialize or deserialize objects.

Injection profile

You now need to define your load testing objective. 

Do you want to simulate production volumes? To understand your cluster’s limits? 

If your cluster is not yet in production and you are in an early stage of development, we can only recommend you to do a capacity test. This type of test will allow you to check the overall behavior of your application. 

{
    setUp(
           kafkaProducer.injectOpen(incrementUsersPerSec(1000)
                   .times(4).eachLevelLasting(60)
                   .separatedByRampsLasting(10)
                   .startingFrom(100.0))
   ).protocols(kafkaProtocol);
}

KafkaSimulation class 

package kafka;

import io.gatling.javaapi.core.*;
import org.apache.kafka.common.header.*;
import org.apache.kafka.common.header.internals.*;
import ru.tinkoff.gatling.kafka.javaapi.protocol.*;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Map;

import static io.gatling.javaapi.core.CoreDsl.*;
import static ru.tinkoff.gatling.kafka.javaapi.KafkaDsl.*;
public class KafkaSimulation extends Simulation{
   public static final String IP_SERVER = System.getProperty("IP_SERVER", "");
   public static final String URL_REGISTRY = System.getProperty("URL_REGISTRY", "");
   public static final String USER_AUTH = System.getProperty("USER_AUTH", "");
   private final KafkaProtocolBuilder kafkaProtocol = kafka()
           .topic("test")
           .properties(
                   Map.of(
                           ProducerConfig.ACKS_CONFIG, "1",

ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",

ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer",

ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG , "org.apache.kafka.common.serialization.StringSerializer")
           );

   private final Headers headers = new RecordHeaders(new Header[]{new RecordHeader("test-header", "value".getBytes())});

   private final ScenarioBuilder kafkaProducer = scenario("Kafka Producer")
       .exec(kafka("Simple Message")
               .send("key","value", headers)
       );

   {
        setUp(

kafkaProducer.injectOpen(incrementUsersPerSec(1000)
                       .times(4).eachLevelLasting(60)
                       .separatedByRampsLasting(10)
                       .startingFrom(100.0))
       ).protocols(kafkaProtocol);
   }
}

Execute the test

We recommend running your test locally to debug it and make sure it works. To do so, run the Engine class or use Maven or Gradle.

You can find the documentation for Maven here:
https://gatling.io/docs/gatling/reference/current/extensions/maven_plugin/#running-your-simulations

And for Gradle here:
https://gatling.io/docs/gatling/reference/current/extensions/gradle_plugin/#running-your-simulations

The next step is to use Gatling Enterprise Cloud to take advantage of key features such as: 

• distributed load testing to inject heavy traffic,
• advanced reporting,
• run history and comparison,
• centralized reporting and access to a multi-tenant solution.
  

To do this, go to https://cloud.gatling.io/ and create an account. Then follow this tutorial to run your first simulation from the cloud.

For Maven: https://gatling.io/docs/gatling/reference/current/extensions/maven_plugin/#api-tokens

For Gradle: https://gatling.io/docs/gatling/reference/current/extensions/gradle_plugin/#api-tokens

Example report from Gatling Enterprise: Here we can see that performance deteriorates at around 60,000 requests per second.

Note: If your Kafka broker cannot be accessed from the Internet, you have two options:

1. Private Locations – If your target environment is private (closed to the Internet).
2. Dedicated IP Addresses – If your target environment is public (open to the Internet) but protected by a firewall.

Both these features will allow you to run your tests in multiple environment configurations. For more information, contact our team.

How to install a Kafka cluster with Docker

If you don’t have access to a Kafka cluster, you can start one on a server using Docker. You can use this Docker Compose configuration file to configure a Kafka cluster.

version: "3"
services:
  kafka:
    container_name: kafka
    image: 'bitnami/kafka:latest'
    ports:
      - 9092:9092
    environment:
      - KAFKA_ENABLE_KRAFT=yes
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
      - KAFKA_BROKER_ID=1
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@127.0.0.1:9093
      - ALLOW_PLAINTEXT_LISTENER=yes

To start this Kafka cluster, create a docker-compose.yml file, copy the contents above and paste it into the file, then run it with:

docker compose up -d

You can then create a test topic with this command:

docker exec -it kafka /opt/bitnami/kafka/bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

To view the messages sent to the topic:

docker exec -it kafka /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic test

To send messages to the topic:

docker exec -it kafka /opt/bitnami/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

Conclusion

We have shown how to use Gatling to test a Kafka cluster with a simple use case. It’s now down to you to develop a more complex scenario fitting your needs. 

We recommend using this method and integrating it into your development cycle using our CI/CD plugins ; it will give you fast and frequent feedback on how your broker is performing. 

Jump in and get started!

Let us know about your project

Used by over 100,000 web applications and e-commerce websites, Gatling solution is put to use in diverse industries including Software, Streaming media, Banking, Insurance, Gaming and Finance.

Related resources

You might also be interested in…