Тестируем Kafka с Testcontainers

Привет, Хабр!

Сегодня мы рассмотрим, как протестировать Kafka с помощью Testcontainers.

Testcontainers — это библиотека, которая из JUnit‑теста запускает Docker‑контейнеры как обычные Java‑объекты. Вы пишете пару строк — а на фоне поднимается полноценная инфраструктура: база, брокер, Redis, что угодно. После теста контейнер гарантированно останавливается, поэтому окружение всегда чистое, а CI не засоряется процессами.

Kafka в Testcontainers запускается теми же двумя строками. Получаем реальный брокер, который ничем не отличается от продакшен‑копии, но живёт ровно столько, сколько идёт тест.

Подключаем зависимости: Gradle и Maven

Если вы на Gradle:

dependencies {     testImplementation platform("org.testcontainers:testcontainers-bom:1.21.3")     testImplementation("org.testcontainers:kafka") }

А если Maven:

<dependencyManagement>   <dependencies>     <dependency>       <groupId>org.testcontainers</groupId>       <artifactId>testcontainers-bom</artifactId>       <version>1.21.3</version>       <type>pom</type>       <scope>import</scope>     </dependency>   </dependencies> </dependencyManagement>  <dependencies>   <dependency>     <groupId>org.testcontainers</groupId>     <artifactId>kafka</artifactId>     <scope>test</scope>   </dependency> </dependencies>

Версия 1.21.3 — на момент написания последняя стабильная.

Первый интеграционный тест с KafkaContainer

Начнём с базового кейса: хочется протестировать, что Kafka работает, продюсер может записать сообщение, а консьюмер прочитать его обратно.

@Testcontainers public class KafkaSmokeTest {      // Объявляем KafkaContainer как @Container — Testcontainers сам поднимет и убьёт его     @Container     static final KafkaContainer kafka = new KafkaContainer(         DockerImageName.parse("apache/kafka-native:3.8.0")     );      @Test     void produceConsume() throws Exception {         // Получаем bootstrap-адрес запущенного брокера         String bootstrap = kafka.getBootstrapServers();          // Конфигурируем Kafka-продюсер         Properties producerProps = new Properties();         producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);         producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());         producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());          // Создаём продюсер и отправляем одно сообщение в demo-topic         KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);         producer.send(new ProducerRecord<>("demo-topic", "my-key", "Hello Kafka!")).get();         producer.close();          // Конфигурируем Kafka-консьюмер         Properties consumerProps = new Properties();         consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);         consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");         consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");         consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());         consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());          // Создаём консьюмера и подписываемся на тот же топик         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);         consumer.subscribe(List.of("demo-topic"));          // Ждём сообщения         ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));          // Проверяем, что пришло именно то сообщение         assertEquals(1, records.count());         assertEquals("Hello Kafka!", records.iterator().next().value());     } }

Контейнер Kafka поднимается автоматически перед тестом, и не нужно ничего запускать вручную. apache/kafka-native:3.8.0 это официальный Kafka‑образ, работающий в KRaft‑режиме (без ZooKeeper).

Вручную создаём продюсера и консьюмера, чтобы максимально контролировать процесс. Проверка assertEquals подтверждает, что сообщение не просто «куда‑то ушло», а дошло до консьюмера.

Что с ZooKeeper и режимами Kafka?

Kafka долгое время зависела от ZooKeeper — для хранения метаданных, регистрации брокеров, и всей магии с кворумами. Но с версии 3.3 Kafka официально переведена в KRaft‑режим (Kafka Raft), где всё хранится внутри самого брокера.

В нашем примере используется образ:

DockerImageName.parse("apache/kafka-native:3.8.0")

Он работает в режиме KRaft по умолчанию. Это значит:

• ZooKeeper не нужен;

• всё конфигурируется проще;

• быстрее стартует;

• меньше зависимостей.

Если вам по какой‑то причине всё‑таки нужен ZooKeeper — например, для старых клиентов или особой топологии, то есть ConfluentKafkaContainer с образом confluentinc/cp-kafka:<до-7.4.0>.

ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:6.2.1");

Но если нет строгих требований используемapache/kafka-native и живем спокойно.

Создание топиков вручную через AdminClient

По дефолту Kafka может создавать топики при первом продюсе. Но это плохая практика, продакшене чаще всего auto.create.topics.enable=false. Поэтому явно создаём нужные топики через AdminClient.

Вот так:

try (AdminClient admin = AdminClient.create(Map.of(         AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers() ))) {     List<NewTopic> topics = List.of(         new NewTopic("input", 1, (short)1),         new NewTopic("primes", 1, (short)1),         new NewTopic("composites", 1, (short)1),         new NewTopic("dlq", 1, (short)1)     );      admin.createTopics(topics).all().get(); }

new NewTopic("имя", партиции, фактор репликации) — в тестах у нас один брокер, так что replicationFactor должен быть строго 1. createTopics(...).all().get() блокирует выполнение до полной регистрации тем на брокере. Без этого возможны гонки.

Если не задать replicationFactor=1, Kafka может начать пытаться найти других брокеров (которых нет) и выбросит TimeoutException.

Пример поинтереснее: маршрутизация по топикам

Читаем из топика числа в виде строк, парсим их, проверяем, простое ли число, и отправляем либо в primes, либо в composites. Если парс не удался — в dlq.

public void routeMessages(KafkaConsumer<String, String> consumer, KafkaProducer<String, String> producer) {     consumer.subscribe(List.of("input"));      while (true) {         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));         for (ConsumerRecord<String, String> record : records) {             try {                 int value = Integer.parseInt(record.value());                 String topic = isPrime(value) ? "primes" : "composites";                 producer.send(new ProducerRecord<>(topic, record.key(), record.value()));             } catch (NumberFormatException ex) {                 producer.send(new ProducerRecord<>("dlq", record.key(), record.value()));             }         }     } }  private boolean isPrime(int n) {     if (n < 2) return false;     for (int i = 2; i <= Math.sqrt(n); i++) {         if (n % i == 0) return false;     }     return true; }

Простой алгоритм, но идеальный для теста. Мы можем:

• отправить 7, получить primes;

• отправить 8, получить composites;

• отправить "abc", получить dlq.

Сетевые listener и взаимодействие между контейнерами

Если есть ещё один контейнер (например, с kcat, или с сервисом) и хочется, чтобы он подключался к Kafka в той же сети:

Network net = Network.newNetwork();  KafkaContainer kafka = new KafkaContainer("apache/kafka-native:3.8.0")     .withNetwork(net)     .withListener("broker:19092"); // будет виден как "broker:19092"  GenericContainer<?> kcat = new GenericContainer<>("confluentinc/cp-kcat:7.9.0")     .withNetwork(net)     .withCreateContainerCmdModifier(cmd -> cmd.withEntrypoint("sh"))     .withCopyToContainer(Transferable.of("7\n8\nabc\n"), "/msgs.txt")     .withCommand("-c", "tail -f /dev/null");  kcat.start(); kafka.start();

Теперь можно из kcat в тесте:

kcat.execInContainer("kcat", "-b", "broker:19092", "-t", "input", "-P", "-l", "/msgs.txt");

Или зачекать результат:

String output = kcat.execInContainer("kcat", "-b", "broker:19092", "-t", "primes", "-C", "-e", "-c", "1").getStdout();

Делитесь своим опытом тестирования Kafka в комментариях, задавайте вопросы, спорьте с решениями — чем больше практических кейсов мы соберём под этой статьёй, тем сильнее станет сообщество. Спасибо, что дочитали!

Приглашаем вас принять участие в серии открытых уроков по Apache Kafka, которые помогут глубже понять ключевые аспекты работы с этой технологией.

30 июля в 19:00 пройдет занятие «Apache Kafka в микросервисной архитектуре — лучшие практики асинхронного обмена». На нем рассмотрим подходы к организации обмена сообщениями в распределенных системах.

13 августа в 18:00 вы сможете ознакомиться с архитектурными паттернами работы с Kafka — на уроке обсудим методы построения устойчивых и масштабируемых решений.

20 августа в 20:00 пройдет урок «Kafka и Clickhouse — как организовать взаимодействие», посвященный интеграции Kafka с аналитической платформой Clickhouse.

А если вы настроены на серьезное обучение, рекомендуем ознакомиться с программой курса по Apache Kafka — на нём максимум практики по работе с Kafka для инженеров данных и разработчиков.