Нагрузочный клиент для тестирования Kafka в формате Avro на Java

Хочу поделиться приёмом, который не раз меня выручал. Есть консьюмер, который читает из Kafka сообщения в формате Avro. Его нужно нагрузочно протестировать. Но схемы Avro у меня нет, бизнес-логику сообщений я не знаю, а ждать, пока соседняя команда соберёт полноценный генератор тестовых данных, нельзя.

Решение оказалось простым: для replay-нагрузки не обязательно понимать Avro. Достаточно один раз сохранить реальное Kafka-сообщение в сыром байтовом виде и отправлять эти байты обратно в топик через ByteArraySerializer. Консьюмер получит ровно тот же wire-format, который получает в проде.

Идея в одной схеме

Важная граница: это не генератор новых бизнес-событий. Это replay-тест, который проверяет, как консьюмер, брокер и downstream выдерживают поток реальных сообщений.

Почему Schema Registry здесь не нужен

Обычно Avro в Kafka живёт вместе со Schema Registry. Сериализатор кладёт в сообщение служебный префикс: magic byte, id схемы и дальше Avro payload. Консьюмер по этому id достаёт схему и десериализует сообщение.

Но если мы не создаём новое Avro-сообщение, а переигрываем уже существующее, нам не нужно заново сериализовать объект. Нам нужно сохранить байты как есть:

Если сохранить payload не как текст, не как JSON и не как «красивое представление», а именно как byte array, консьюмер не заметит разницы.

Конфигурируем продюсер

Нам нужен обычный producer, у которого ключ и значение — byte[]. Конфигурацию удобнее держать явно в Java-коде: для нагрузочного клиента так проще видеть, какие параметры реально участвуют в тесте.

@Configuration
public class LoadProducerConfig {

    @Value("${app.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<byte[], byte[]> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

        // Если кластер за SASL/SSL, сюда добавляются обычные security.* и sasl.* свойства.
        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    public KafkaTemplate<byte[], byte[]> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

Здесь принципиально нет KafkaAvroSerializer и schema.registry.url. Мы не сериализуем Avro, а отправляем готовую последовательность байт.

Для сравнения, если бы мы генерировали новые Avro-сообщения, конфигурация была бы другой:

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

Но это уже другой класс задачи.

Как достать сырые байты

Самое важное — не испортить исходное сообщение при сохранении. Нужен инструмент, который умеет выгрузить record value в binary form.

Надёжный вариант — написать маленький consumer, который читает одно сообщение и пишет record.value() в файл.

byte[] payload;

try (InputStream in = getClass().getResourceAsStream("/sample-message.bin")) {
    if (in == null) {
        throw new IllegalStateException("sample-message.bin not found");
    }
    payload = in.readAllBytes();
}

Файл лучше назвать .bin, а не .avro: внутри лежит не чистый Avro-файл, а Kafka wire-format сообщения.

Создаём нагрузку

Ниже минимальный вариант: задаём количество сообщений и целевой RPS, отправляем payload в цикле, считаем задержку callback-а.

int total = 100_000;
int targetRps = 1_000;
long intervalNanos = 1_000_000_000L / targetRps;

long startedAt = System.nanoTime();
Queue<Long> latencies = new ConcurrentLinkedQueue<>();
AtomicInteger errors = new AtomicInteger();

for (int i = 0; i < total; i++) {
    long sentAt = System.nanoTime();

    kafkaTemplate.send("my-topic", payload)
            .whenComplete((result, error) -> {
                latencies.add(System.nanoTime() - sentAt);
                if (error != null) {
                    errors.incrementAndGet();
                }
            });

    long nextSendAt = startedAt + (long) i * intervalNanos;
    long sleepNanos = nextSendAt - System.nanoTime();
    if (sleepNanos > 0) {
        LockSupport.parkNanos(sleepNanos);
    }
}

kafkaTemplate.flush();

double seconds = (System.nanoTime() - startedAt) / 1e9;
List<Long> sorted = new ArrayList<>(latencies);
Collections.sort(sorted);

System.out.printf("sent=%d errors=%d duration=%.1fs throughput=%.0f msg/s%n",
        total, errors.get(), seconds, total / seconds);
System.out.printf("callback latency p50=%.1fms p95=%.1fms p99=%.1fms%n",
        percentile(sorted, 50) / 1e6,
        percentile(sorted, 95) / 1e6,
        percentile(sorted, 99) / 1e6);

static long percentile(List<Long> sorted, int p) {
    if (sorted.isEmpty()) {
        return 0;
    }
    int index = (int) Math.ceil(p / 100.0 * sorted.size()) - 1;
    return sorted.get(Math.max(0, Math.min(index, sorted.size() - 1)));
}

Что обязательно мерить

Если смотреть только на «сколько сообщений отправил producer», тест легко обманет. Producer может радостно отправлять, а консьюмер уже тонет.

Минимальный набор метрик:

Обычно я гоняю несколько ступеней: 100, 500, 1000, 2000 msg/s и смотрю, где начинается нелинейный рост lag или p99.

Когда нужны настоящие Avro-сообщения

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

Настоящий Avro + Schema Registry нужен, если:

Тогда нужен полноценный генератор данных: Avro-классы, KafkaAvroSerializer, Schema Registry и сценарий изменения полей.

Чем ещё можно нагружать

Мой способ хорош тем, что его можно собрать за вечер. Но он не единственный.

В моём случае replay-client занимает нишу между штатным perf-test и полноценным генератором бизнес-событий.

Взгляд из 2026

С 2022 года Kafka-инфраструктура заметно изменилась. К 2026 году актуальная ветка Apache Kafka уже 4.x; в 4.0 ZooKeeper окончательно убрали, а KRaft стал единственным режимом управления кластером. Для эксплуатации это важно, но сам трюк с ByteArraySerializer от этого не меняется.

Spring-экосистема тоже обновилась: Spring for Apache Kafka 4.x продолжает давать KafkaTemplate как высокоуровневую обёртку для отправки сообщений. Для нашего replay-клиента это значит одно: код можно держать на современных версиях, но проверять надо не только producer API, а ещё и настройки безопасности, compression, batching и лимиты брокера.

Что я бы добавил в 2026 году:

Итог

Если задача — быстро нагрузить консьюмер на реальном Avro-потоке, не зная схемы, не надо насильно разбираться в Avro. Сохраните одно сообщение как bytes, отправляйте его через ByteArraySerializer, меряйте producer latency, consumer lag и ошибки downstream.

Главное — честно помнить границы метода: replay байтов проверяет устойчивость обработки, но не заменяет генератор валидных бизнес-событий.

Ссылки

• Apache Kafka documentation

• Apache Kafka 4.0 release announcement

• Spring for Apache Kafka reference

• Confluent Schema Registry serializers and wire format

• Confluent producer configuration reference