Active users grow with my custom protocol implementation and constantUsersPerSec

Muneyuki_Noguchi · January 28, 2015, 9:37pm

Hello,

I implemented an Apache Kafka 0.8 protocol support using Gatling 2.1.3:

However, the graph in an HTML report shows that active users grow even when I use constantUsersPerSec as the attached image file
though the number of threads in a java process doesn’t grow so much during a stress test.

The simulation file I used is as follows:

github.com

mnogu/gatling-kafka/blob/0.0.2/src/test/scala/com/github/mnogu/gatling/kafka/test/BasicSimulation.scala

package com.github.mnogu.gatling.kafka.test

import io.gatling.core.Predef._
import org.apache.kafka.clients.producer.ProducerConfig
import scala.concurrent.duration._

import com.github.mnogu.gatling.kafka.Predef._

class BasicSimulation extends Simulation {
  val kafkaConf = kafka
    .topic("test")
    .properties(
      Map(
        ProducerConfig.ACKS_CONFIG -> "1",
        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG ->  "localhost:9092"))

  val scn = scenario("Kafka Test")
    .exec(kafka("request").send("foo"))
  // You can also use feeder
  //

This file has been truncated. show original

And the code to send requests to Apache Kafka 0.8 is as follows:

github.com

mnogu/gatling-kafka/blob/0.0.2/src/main/scala/com/github/mnogu/gatling/kafka/action/KafkaRequestAction.scala

package com.github.mnogu.gatling.kafka.action

import akka.actor.ActorRef

import com.github.mnogu.gatling.kafka.config.KafkaProtocol
import com.github.mnogu.gatling.kafka.request.builder.KafkaAttributes
import io.gatling.core.action.{Failable, Interruptable}
import io.gatling.core.result.message.{KO, OK}
import io.gatling.core.result.writer.DataWriterClient
import io.gatling.core.session._
import io.gatling.core.util.TimeHelper.nowMillis
import io.gatling.core.validation.Validation
import org.apache.kafka.clients.producer._

import scala.collection.JavaConverters._

object KafkaRequestAction extends DataWriterClient {
  def reportUnbuildableRequest(
      requestName: String,
      session: Session,

This file has been truncated. show original

Am I doing something wrong? If so, how to fix bugs in my code?

Thanks in advance.

slandelle · January 29, 2015, 12:24am

Hi,

Your issue is that you’re blocking on the send Futures: https://github.com/mnogu/gatling-kafka/blob/master/src/main/scala/com/github/mnogu/gatling/kafka/action/KafkaRequestAction.scala#L61

As a results, you have a thread starvation, and users pile up in the starting queue without getting a chance to run.

As KafkaProducer.send returns regular Futures and not CompletableFutures so you can react on completion instead of blocking, you should use instead the version that takes Callback:
https://github.com/apache/kafka/blob/0.8.1/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L212

Get it?

Cheers,

Muneyuki_Noguchi · January 29, 2015, 10:42pm

The number of active users still grows with a callback, but the growth rate becomes smaller:

github.com

mnogu/gatling-kafka/blob/0.0.3/src/main/scala/com/github/mnogu/gatling/kafka/action/KafkaRequestAction.scala#L58

      
        
              session: Session): Validation[Unit] = {
            
            
payload(session).map { resolvedPayload =>
              val producer = new KafkaProducer(kafkaProtocol.properties.asJava)
              val record = new ProducerRecord(
                kafkaProtocol.topic, resolvedPayload.getBytes)
            
            
  val requestStartDate = nowMillis
              val requestEndDate = nowMillis
            
            
  // send the request
              producer.send(record, new Callback() {
                override def onCompletion(m: RecordMetadata, e: Exception): Unit = {
                  val responseStartDate = nowMillis
                  val responseEndDate = nowMillis
            
            
      // log the outcome
                  writeRequestData(
                    session,
                    requestName,
                    requestStartDate,

2015年1月29日木曜日 9時24分13秒 UTC+9 Stéphane Landelle:

slandelle · January 30, 2015, 6:39am

Are you sure creating and closing a producer on each event is the proper pattern? Is this how Kafka is supposed to be used? Shouldn’t such component be reused and shared?

Muneyuki_Noguchi · January 31, 2015, 2:07am

Thank you for your advice.
The issue in my implementation is solved.

https://github.com/mnogu/gatling-kafka/commit/5c23ed47629b9187dd06e2bf94ea47c174d0e3e0

https://github.com/mnogu/gatling-kafka/blob/0.0.4/src/main/scala/com/github/mnogu/gatling/kafka/action/KafkaRequestActionBuilder.scala#L22
https://github.com/mnogu/gatling-kafka/blob/0.0.4/src/main/scala/com/github/mnogu/gatling/kafka/action/KafkaRequestAction.scala

2015年1月30日金曜日 15時39分35秒 UTC+9 Stéphane Landelle:

slandelle · January 31, 2015, 7:51am

Great!

Topic		Replies	Views
"Active Users" metric growth extremely during "constantUsersPerSec" load strategy. Gatling (Open-Source)	3	90	September 25, 2015
Simulate a number of active users Gatling (Open-Source)	3	124	May 13, 2017
Gatling Kafka extensions : Consumers Gatling (Open-Source)	3	109	March 24, 2018
constantUsersPerSec() during () Gatling (Open-Source)	2	108	March 8, 2019
constant users per sec Gatling (Open-Source)	3	134	December 9, 2016

Active users grow with my custom protocol implementation and constantUsersPerSec

Related Topics