Нагрузочное тестирование Apache Kafka с использованием JMeter

Проще говоря, Apache Kafka — это гибрид распределенной базы данных и очереди сообщений. Для обработки терабайтов информации многие крупные компании используют ее. Также, благодаря своим особенностям, Кафка широко популярна. Например, такая компания, как LinkedIn, использует ее для потоковой передачи данных о пользовательской активности, а такая компания, как Netflix, использует ее для сбора и буферизации данных для последующих систем, таких как Elasticsearch, Amazon EMR, Mantis и многих других.

Кроме того, давайте рассмотрим некоторые особенности Kafka, которые важны для нагрузочного тестирования Kafka:

• По умолчанию длительное время хранения сообщений — неделя.
• Благодаря последовательному вводу / выводу, высокая производительность.
• Также удобна кластеризация.
• Для репликации и распределения очередей по кластеру, высокая доступность данных благодаря возможности.
• Помимо передачи данных, он также может обрабатывать с помощью потокового API.

Как известно, для работы с очень большим объемом данных мы используем Kafka. Следовательно, во время нагрузочного тестирования Kafka с JMeter обратите внимание на следующие аспекты:

• Если мы будем постоянно записывать данные на диск, это будет влиять на производительность сервера. Потому что он достигнет состояния отказа в обслуживании, если его недостаточно.
• Тем не менее, распределение разделов и количество брокеров также влияет на использование сервисного потенциала.
• Однако все становится еще сложнее, пока мы используем функцию репликации. Причиной этого является то, что его обслуживание требует еще больше ресурсов, и случай, когда брокеры отказываются принимать сообщения, становится еще более вероятным.

Хотя существует вероятность того, что данные могут потеряться, когда они обрабатываются в таких огромных количествах, даже если большинство процессов автоматизировано. Следовательно, мы можем сказать, что тестирование этих сервисов очень важно; Кроме того, важно иметь возможность генерировать правильную нагрузку.

Убедитесь, что нагрузочное тестирование Apache Kafka будет установлено в Ubuntu, чтобы продемонстрировать это. Кроме того, мы будем использовать   плагин Pepper-Box в качестве производителя, поскольку он более удобен для работы с генерацией сообщений, чем  kafkameter  . Однако ни один плагин не обеспечивает реализацию потребителем, поэтому мы должны реализовать потребителя самостоятельно. И мы собираемся использовать  сэмплер  JSR223 для этого. Теперь перейдем к нагрузочному тестированию Kafka.

Настройка производителя: Pepper-Box

Теперь, чтобы установить плагин, нам нужно скомпилировать этот исходный код или скачать файл jar. Далее нам нужно поместить его в папку lib / ext и перезапустить JMeter.

По сути, есть три элемента этого плагина:

• Pepper-Box PlainText Config — позволяет создавать текстовые сообщения в соответствии с указанным шаблоном.
• Сериализованная конфигурация Pepper-Box — эта конфигурация позволяет нам создавать сообщение, которое является сериализованным Java-объектом.
• PepperBoxKafkaSampler — предназначен для отправки сообщений, созданных предыдущими элементами.

Давайте подробно изучим все эти конфигурации для нагрузочного тестирования Kafka:

Конфигурация Pepper-Box PlainText

Выполните следующие действия, чтобы добавить этот элемент. Сначала перейдите в группу  Thread -> Добавить -> Элемент конфигурации -> Настройка Pepper-Box PlainText

Элемент конфигурации Pepper-Box PlainText в Kafka Load Testing имеет два поля:

Ключ заполнителя сообщения

Хотя мы хотим использовать шаблон из этого элемента, это ключ, который необходимо указать в PepperBoxKafkaSampler.

Шаблон схемы

Это шаблон сообщения, в котором мы можем использовать переменные и функции JMeter, которые также могут подключать функции. Однако эта структура сообщения может быть любой, от простого текста до JSON или XML.

Сериализованная конфигурация Pepper-Box

Теперь выполните несколько шагов, чтобы добавить этот элемент. Во-первых, перейдите в  группу потоков -> Добавить -> Элемент конфигурации -> Сериализованная конфигурация Pepper-Box

However, this element has a field for the key and the Class Name field, which is intended for specifying the Java class. Makes sure that the jar file with the class, must be placed in the lib/ext folder. Hence, the fields with its properties will appear below just after it is specified, and also it is possible to assign desired values to them now. Although, here again, we repeated the message from the last element, it will be a Java object this time.

PepperBoxKafkaSampler

Further, in order to add this element, follow these steps. First, go to Thread group -> Add -> Sampler -> Java Request. Afterward, select com.gslab.pepper.sampler.PepperBoxKafkaSampler from the drop-down list. 

Basically, there are the following settings of this element:

• bootstrap.servers/zookeeper.servers

Addresses of brokers/ zookeepers in the format of broker-ip-1: port, broker-ip-2: port, etc.

• kafka.topic.name

It is the name of the topic for message publication.

• key.serializer

However, it is a class for key serialization. If there is no key in the message, leave it unchanged.

• value.serializes

It is a class for message serialization. The field remains unchanged, for simple text. We need to specify “com.gslab.pepper.input.serialized.ObjectSerializer,” when using Pepper-Box Serialized Config.

• compression.type — Basically, this is a type of message compression (none/gzip/snappy/lz4)
• batch.size — This is the largest message size.
• linger.ms — This is considered the message waiting time.
• buffer.memory — This is the producer’s buffer size.
• acks — This is the quality of service (-1/0/1 – delivery is not guaranteed/the message will surely be delivered/the message will be delivered once).
• receive.buffer.bytes/send.buffer.bytes — This is the size of the TCP send/receive buffer. -1 – use the default OS value.
• security.protocol — This is the encryption protocol (PLAINTEXT/SSL/SASL_PLAINTEXT/ SASL_SSL).
• message.placeholder.key — This is the message key, which was specified in the previous elements.
• kerberos.auth.enabled,java.security.auth.login.config,java.security.krb5.conf,sasl.kerberos.service.name — These all are a field group responsible for the authentication.

Generally, we can add additional parameters using the prefix _ before the name, if necessary.
For example, _ssl.key.password.

Configuring the Consumer

As we have a producer that creates the largest load on the server, the service has to deliver messages too. Hence, to more accurately reproduce the situations, we should also add consumers. Moreover, we can also use it to check whether all consumer messages have been delivered.
For example, 
Let’s take the following source code and briefly touch upon its steps for a better understanding of Kafka Load Testing:

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("group.id", group);
 props.put("enable.auto.commit", "true");
 props.put("auto.commit.interval.ms", "1000");
 props.put("session.timeout.ms", "30000");
 props.put("key.deserializer",
 "org.apache.kafka.common.serializa-tion.StringDeserializer");
 props.put("value.deserializer",
 "org.apache.kafka.common.serializa-tion.StringDeserializer");
 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
 consumer.subscribe(Arrays.asList(topic));
 System.out.println("Subscribed to topic " + topic);
 int i = 0;
 while (true) {
 ConsumerRecords<String, String> records = con-sumer.poll(100);
 for (ConsumerRecord<String, String> record : records)
 System.out.printf("offset = %d, key = %s, value = %s\n",
 record.offset(), record.key(), record.value());
 }

Updated part:
 long t = System.currentTimeMillis();
 long end = t + 5000;
 f = new FileOutputStream(".\\data.csv", true);
 p = new PrintStream(f);
 while (System.currentTimeMillis()<end)
 {
 ConsumerRecords<String, String> records = consumer.poll(100);
 for (ConsumerRecord<String, String> record : records)
 {
 p.println( "offset = " + record.offset() +" value = " + record.value());
 }
 consumer.commitSync();
 }
 consumer.close();
 p.close();
 f.close();