Kafka Consumer в тестовой автоматизации: архитектурный разбор

Привет, Хабр! На связи Егор Лаптев — QA Fullstack Java в SENSE на проекте крупного российского банка.

End-to-end тесты UI проверяют только то, что видно на экране: кнопка нажалась, форма открылась, данные отобразились. Но в распределённых системах значительная часть бизнес-логики уходит за кадр — в асинхронные события, которые летят через Kafka. Если событие не дошло до топика или пришло с неверным payload, пользователь этого не увидит, а бизнес-процесс сломается.

В этой статье расскажу, как мы научились проверять Kafka-события прямо в автотестах — без Kafka UI, kcat и обёрточных сервисов, одной зависимостью и так, чтобы это работало в корпоративной сети с SSL. Покажу архитектуру коннектора, разберу три основных проблемы (SSL-сертификаты, конфликты consumer group и асинхронные тайминги), и поделюсь моделью работы инженера в связке с AI-агентом.

Кому будет полезно: QA-инженерам и специалистам по автоматизации, которые тестируют распределённые системы и хотят проверять Kafka-события прямо в автотестах; и инженерам, которым интересна практическая модель работы в связке с AI-агентом — как она ускоряет разработку тестовой инфраструктуры.

Контекст проекта

Мы столкнулись с тем, что тесты показывали зелёный, а в продакшене сыпались инциденты, потому что никто не проверял, доходят ли события до Kafka. Готового решения, которое можно было бы просто встроить в автотесты, не нашлось: всё либо тянет за собой инфраструктуру, либо не работает в корпоративных сетях с SSL.

Задача была в том, чтобы снизить зависимость от внешних источников и ручных проверок. Раньше верификация Kafka-событий была ручной: тестировщик запускал тест, шёл в Kafka UI или консоль, искал сообщение, проверял payload. Это медленно, ненадёжно и не масштабируется. Мы хотели, чтобы проверка событий была такой же автоматизированной, как и проверка UI.

При этом сам коннектор — лишь часть более широкого подхода. Он не появился бы сам по себе: его сделал возможным принцип, при котором инженер по автоматизации не ограничивается ролью «пишу тесты по ТЗ», а ищет точки входа — места, где ручной труд можно заменить автоматизацией, а автоматизацию усилить AI. AI-агент при этом встроен в процессы как полноценный участник команды: от исследования и генерации кода до отладки и ревью. Коннектор — один из результатов такого подхода.

Что получилось: полностью самостоятельный коннектор без обёрточных сервисов. Никакого Kafka UI, Confluent Control Center, Kafka REST Proxy или kcat — только прямое подключение через kafka-clients. Одна зависимость в build.gradle, никаких контейнеров и прокси. Работает локально, в CI/CD, в Docker — достаточно указать bootstrapServers и креды.

Что мы получили в цифрах:

Какие боли решает коннектор

Дальше — детальный разбор каждой боли и решения.

Архитектура

src/test/java/com/example/├── client/kafka/│   ├── KafkaConsumerClientjava — ядро: consumer, poll, фильтрация |── CustomSslEngineFactory.java   — SSL-обход для корпоративных сертификатов├── steps/│   └── KafkaBddSteps.java        — BDD-шаги (Given/When/Then)└── utils/└── ScenarioContext.java       — Singleton-хранилище состояния src/test/resources/├── configuration/config.properties— Kafka-конфигурация (camelCase)└── .env                           — Kafka-конфигурация (UPPER_SNAKE_CASE)

Зависимость (build.gradle):

testImplementation ‘org.apache.kafka:kafka-clients:3.7.0’

Consumer-обёртка — ядро коннектора

Два конструктора

Три режима подключения

Ключевая разница: subscribe() участвует в consumer group — offset фиксируется. assign*() — прямое назначение партиций, без влияния на consumer group offset.

Чтение сообщений

Все прочитанные сообщения накапливаются во внутреннем буфере — это позволяет фильтровать их позже.

Фильтрация накопленных сообщений

findMessagesContaining(substring)findMessagesByKey(key)getConsumedMessages()clearConsumedMessages()

Свойства Consumer

Consumer настраивается со следующими ключевыми свойствами: чтение с начала топика (earliest), ручной коммит offset после обработки и лимит в 100 записей за один poll. Десериализаторы ключей и значений — строковые.

auto.offset.reset=earliestenable.auto.commit=falsemax.poll.records=100key.deserializer=StringDeserializervalue.deserializer=StringDeserializer

Жизненный цикл

Класс реализует AutoCloseable — закрытие consumer’а гарантировано в хуке после сценария:

@Afterpublic void closeKafkaConsumer() {Object consumer = context.getData("consumerKey");if (consumer instanceof KafkaConsumerWrapper) {    ((KafkaConsumerWrapper) consumer).close();}}

CustomSslEngineFactory — метод configure()

public class CustomSslEngineFactory implements SslEngineFactory {    private SSLContext sslContext;    @Override    public void configure(Map<String, ?> configs) {        try {            TrustManager[] trustAllCerts = new TrustManager[]{                new X509TrustManager() {                    public X509Certificate[] getAcceptedIssuers() {                        return new X509Certificate[0];                    }                    public void checkClientTrusted(X509Certificate[] certs,                                                    String authType) {}                    public void checkServerTrusted(X509Certificate[] certs,                                                    String authType) {}                }            };            sslContext = SSLContext.getInstance("TLS");            sslContext.init(null, trustAllCerts, new SecureRandom());        } catch (Exception e) {            throw new RuntimeException(                    "Не удалось инициализировать CustomSslEngineFactory", e);        }    }    @Override    public SSLEngine createClientSslEngine(String peerHost, int peerPort,                                            String endpointIdentification) {        SSLEngine engine = sslContext.createSSLEngine(peerHost, peerPort);        engine.setUseClientMode(true);        engine.setSSLParameters(new SSLParameters());        return engine;    }// shouldBeRebuilt, reconfigurableConfigs, keystore, truststore, close — бойлерплейт}

KafkaConsumerClient — подключение и poll-цикл

public class KafkaConsumerClient implements AutoCloseable {    private KafkaConsumer<String, String> consumer;    private final List<ConsumerRecord<String, String>> consumedMessages = new ArrayList<>();    // ... конструкторы, поля, buildConsumerProperties() ...    @Step("Подписка на Kafka топик: {topic}")    public void subscribe(String topic) {        consumer = new KafkaConsumer<>(buildConsumerProperties());        consumer.subscribe(Collections.singletonList(topic));        consumer.poll(Duration.ofMillis(3000));    }    @Step("Назначение партиции топика: {topic}, партиция: {partition}")    public void assignPartition(String topic, int partition) {        consumer = new KafkaConsumer<>(buildConsumerProperties());        TopicPartition tp = new TopicPartition(topic, partition);        consumer.assign(Collections.singletonList(tp));        consumer.seekToBeginning(Collections.singletonList(tp));    }    @Step("Назначение всех партиций топика: {topic}")    public void assignAllPartitions(String topic) {        consumer = new KafkaConsumer<>(buildConsumerProperties());        List<TopicPartition> partitions = consumer.partitionsFor(topic).stream()                .map(pi -> new TopicPartition(topic, pi.partition()))                .toList();        consumer.assign(partitions);        consumer.seekToBeginning(partitions);    }    @Step("Чтение сообщений из Kafka топика: {timeoutSeconds} сек")    public List<ConsumerRecord<String, String>> pollMessages(int timeoutSeconds) {        List<ConsumerRecord<String, String>> result = new ArrayList<>();        long endTime = System.currentTimeMillis() + timeoutSeconds * 1000L;        while (System.currentTimeMillis() < endTime) {            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));            for (ConsumerRecord<String, String> record : records) {                result.add(record);                consumedMessages.add(record);            }            if (!records.isEmpty()) {                consumer.commitSync();            }        }        return result;    }    @Step("Чтение одного сообщения из Kafka топика")    public ConsumerRecord<String, String> pollOneMessage(int timeoutSeconds) {        long endTime = System.currentTimeMillis() + timeoutSeconds * 1000L;        while (System.currentTimeMillis() < endTime) {            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));            if (!records.isEmpty()) {                ConsumerRecord<String, String> record = records.iterator().next();                consumedMessages.add(record);                consumer.commitSync();                return record;            }        }        return null;    }    // findMessagesContaining, findMessagesByKey, getConsumedMessages,    // clearConsumedMessages, close — опущены для краткости}

KafkaSteps — примеры BDD-шагов

public class KafkaSteps {    @Дано("подключение к Kafka топику {string}")    public void subscribeToTopic(String topic) {        getConsumer().subscribe(topic);    }    @Дано("назначение всех партиций Kafka топика {string}")    public void assignAllPartitions(String topic) {        getConsumer().assignAllPartitions(topic);    }      @Когда("прочитаны сообщения из Kafka топика в течение {int} секунд")    public void pollMessages(int timeoutSeconds) {        List<ConsumerRecord<String, String>> messages =                getConsumer().pollMessages(timeoutSeconds);        ScenarioContext.getInstance().setData("kafkaMessages", messages);    }    @Тогда("в Kafka топике есть сообщения")    public void verifyMessagesExist() {        @SuppressWarnings("unchecked")        List<ConsumerRecord<String, String>> messages = (List<ConsumerRecord<String, String>>)                ScenarioContext.getInstance().getData("kafkaMessages");        assertNotNull(messages);        assertFalse(messages.isEmpty(), "В Kafka топике нет сообщений");    }      @Тогда("сообщение из Kafka топика содержит текст {string}")    public void verifyMessageContainsText(String expectedText) {        @SuppressWarnings("unchecked")        List<ConsumerRecord<String, String>> messages = (List<ConsumerRecord<String, String>>)                ScenarioContext.getInstance().getData("kafkaMessages");        boolean found = messages.stream()                .anyMatch(r -> r.value() != null && r.value().contains(expectedText));        assertTrue(found, "Ни одно сообщение не содержит текст: " + expectedText);    }      @Тогда("значение последнего сообщения из Kafka топика сохранено в контекст как {string}")    public void saveLastMessageValueToContext(String key) {        ConsumerRecord<String, String> message = (ConsumerRecord<String, String>)                ScenarioContext.getInstance().getData("kafkaLastMessage");        assertNotNull(message, "Сообщение не было прочитано");        ScenarioContext.getInstance().setData(key, message.value());    }}

1. SSL-сертификаты: когда Java отказывается подключаться

Kafka-кластер в корпоративной среде работает через SASL_SSL с внутренними сертификатами, которые не входят в стандартный Java truststore. При первой попытке подключения consumer падал с SSLHandshakeException — Java просто не доверяла корпоративному CA.

Что пробовали:

• Импорт сертификата в JVM truststore (keytool -import) — работает на одной машине, но ломается в CI/CD, где агенты динамические и truststore нужно настраивать на каждом раннере;

• Передача truststore через ssl.truststore.location — требует хранения JKS-файла и его синхронизации между средами;

• Обёрточные сервисы (Kafka REST Proxy, kcat) — тянут за собой инфраструктуру, контейнеры, поддержку.

Что сделали: написали собственную CustomSslEngineFactory — полностью уникальную реализацию интерфейса SslEngineFactory из kafka-clients. Это не обёртка над готовой библиотекой и не костыль с TrustManager[] на уровне HttpsURLConnection. Это кастомная фабрика SSL-движков, которая инжектируется напрямую в Kafka consumer через конфигурацию:

props.put(«ssl.engine.factory.class», «com.example.client.kafka.CustomSslEngineFactory»);

Фабрика создаёт SSLContext с TrustManager, который принимает любой сертификат, и оборачивает его в SSLEngine через createSSLEngine(String peerHost, int peerPort). Без этого коннектор физически не мог бы работать в корпоративной сети — ни один стандартный подход не подходил для динамических CI/CD-агентов.

⚠️ Внимание: данный подход небезопасен для продакшена. TrustAllSslEngineFactory отключает проверку SSL-сертификатов, что создаёт уязвимость к MITM-атакам. Мы применяем это решение исключительно внутри изолированного тестового контура (Staging/QA), где:

• Kafka-кластер недоступен извне корпоративной сети;

• все подключённые сервисы — тестовые стенды, не обрабатывающие реальные данные клиентов;

• доступ к кластеру ограничен SASL/SCRAM аутентификацией.

В продакшене необходимо использовать стандартную проверку сертификатов через JKS truststore. Если ваша тестовая среда позволяет импортировать корпоративный CA в JVM truststore — это предпочтительный вариант. Наш подход — компромисс для динамических CI/CD-агентов, где truststore невозможно настроить статически.

Здесь AI-агент проявил себя особенно ярко: реализация SslEngineFactory — это не стандартный паттерн, который можно нагуглить. Потребовалось понимание внутренних механизмов kafka-clients и Java SSL API. AI-агент сгенерировал корректную реализацию с первого раза, включая все методы интерфейса — createSSLEngine, createClientSslEngine, createServerSslEngine, shouldBeRebuilt и другие. Это тот случай, когда AI не просто «помог с кодом», а решил задачу, которая без него заняла бы дни изучения документации.

2. Consumer Group конфликты: когда автотесты ломают тестовую среду

Первая версия коннектора использовала subscribe() с дефолтным groupId из конфигурации. На первый взгляд — всё работало. Но при параллельном запуске тестов обнаружилось, что реальные сервисы-консьюмеры на тестовой среде перестали получать сообщения. Причина: автотесты с тем же groupId участвовали в rebalance и «отбирали» партиции у рабочих сервисов.

Что сделали:

1. Ввели правило: автотесты всегда используют уникальный groupId, не пересекающийся с группами реальных сервисов;

2. Добавили второй режим подключения — assignAllPartitions() — для чтения с начала топика без участия в consumer group вообще. Никакого rebalance, никаких конфликтов;

3. Добавили BDD-шаг с кастомным groupId для параллельных прогонов — каждый сценарий получает свою группу.

Четыре правила работы с consumer groups в автотестах:

Правило №1: всегда выделяйте отдельную consumer group для автотестов. В конфигурации автотестов должен быть уникальный groupId, который не пересекается с группами реальных сервисов:

kafkaGroupId=some-topic.ui-autotests

Правило №2: если нужно читать с начала топика — используйте assignAllPartitions(). Режим assign не участвует в consumer group rebalance и не влияет на другие консьюмеры.

Правило №3: подключайтесь к Kafka ДО триггерного действия. Consumer должен успеть подписаться и пройти rebalance до того, как тестируемое действие отправит сообщение. Иначе сообщение может быть прочитано другим участником группы или пропущено:

1. [шаг подключения к топику…]

2. [триггерное действие, порождающее событие…]

3. [шаг чтения сообщений с таймаутом…]

Правило №4: для параллельных тестов — уникальный groupId на каждый сценарий. Если тесты запускаются параллельно и используют один groupId, они будут конкурировать за партиции внутри одной consumer group. Решение — генерировать уникальный groupId для каждого сценария.

Эта проблема — классический пример того, как важно не просто написать код, а понять контекст: как система работает в целом, кто ещё читает из топика, что произойдёт при параллельном запуске. Инженер, который ищет точки входа для улучшения процессов, видит такие риски до того, как они превращаются в инциденты.

3. Асинхронные тайминги: когда sleep() не работает

Сообщение в Kafka появляется не мгновенно после действия в системе — между триггером и появлением события в топике проходит от нескольких секунд до минуты. Первая попытка использовать Thread.sleep(10000) привела к двум проблемам: иногда 10 секунд не хватало (тест падал), иногда сообщение появлялось за 2 секунды (тест ждал впустую 8 секунд, замедляя прогон).

Что сделали: реализовали poll-цикл с настраиваемым таймаутом. Consumer опрашивает топик каждую секунду (poll(Duration.ofSeconds(1))) до тех пор, пока не получит сообщение или не истечёт таймаут. Сообщение захватывается в момент появления — без лишних ожиданий и без пропусков:

while (elapsed < timeoutSeconds) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));if (!records.isEmpty()) {    break;}elapsed++;}

4. Ручная верификация → BDD-шаги

Раньше тестировщик после запуска теста шёл в Kafka UI или консоль, искал сообщение, проверял payload руками. Это медленно, ненадёжно и не масштабируется. Коннектор заменяет весь этот процесс на BDD-шаги, которые выполняются прямо в сценарии.

Подключение (Given / Дано)

Чтение (When / Когда)

Проверки (Then / Тогда)

Паттерны использования

Паттерн 1: Подписка ДО триггерного действия

→ Выполняются предусловия → Подключение к Kafka → Выполняется триггерное действие → Проверка сообщения

Ключевая идея: consumer подключается до действия, которое порождает событие, чтобы гарантированно захватить сообщение.

Паттерн 2: Подписка ПОСЛЕ триггерного действия

→ Выполняются предусловия → Выполняется триггерное действие → Подключение к Kafka → Дополнительное действие → Проверка сообщения

Когда использовать: когда событие отправляется не сразу, а после дополнительного действия пользователя.

Паттерн 3: Негативная проверка — сообщений быть не должно

→ Выполняются предусловия → Подключение к Kafka → Выполняется действие → Проверка что сообщений НЕТ

Негативная проверка: определённые действия не должны порождать событий в Kafka.

5. Внешние зависимости → прямое подключение

Kafka REST Proxy, kcat, Confluent Control Center — все эти инструменты тянут за собой инфраструктуру, контейнеры, поддержку. В корпоративной среде это особенно критично: каждый новый сервис — это заявка на доступ, настройка SSL, синхронизация версий.

Коннектор использует прямое подключение через kafka-clients — одна зависимость в build.gradle, никаких контейнеров и прокси. Работает локально, в CI/CD, в Docker — достаточно указать bootstrapServers и креды.

Это даёт:

• Минимальные зависимости — одна строка в build.gradle;

• Максимальный контроль — фильтрация, таймауты, режимы подключения — всё настраивается программно;

• Работает везде — локально, в CI/CD, в Docker;

• Интеграция с отчётами — каждое действие логируется, в отчёте виден полный путь сообщения от топика до проверки.

Интеграция с отчётами

Каждый ключевой метод аннотирован @Step и добавляет вложения в отчёт:

@Step("Подписка на Kafka топик: {topic}")public void subscribe(String topic) { ...Allure.addAttachment("Kafka подписка", "Подписка на топик: " + topic + ", groupId: " + groupId);} @Step("Чтение одного сообщения из Kafka топика")public ConsumerRecord<String, String> pollOneMessage(int timeoutSeconds) { ...Allure.addAttachment("Kafka сообщение",    "Топик: " + record.topic() + ", партиция: " + record.partition() +    ", offset: " + record.offset() + ", key: " + record.key() + ", value: " + record.value());}

В отчёте видно: какой топик, какой groupId, сколько сообщений прочитано, и полное содержание каждого сообщения.

Контекст сценария — как связываются компоненты

Контекст сценария — Singleton на основе HashMap<String, Object>. Kafka использует 3 ключа:

Скорость разработки: неделя за один день

Оценка ручной разработки такого коннектора «с нуля» — около недели: изучение SASL/SCRAM-конфигурации, реализация SslEngineFactory, написание consumer с poll-циклом, BDD-шаги, интеграция с контекстом сценария и хуками, отладка SSL-соединений, аннотации для отчётности.

С AI-ассистентом рабочее решение, прошедшее ревью, было готово за один рабочий день. Формат работы:

1. Инженер определяет архитектурные решения: режимы подключения, модель данных, конфигурация, сценарии использования;

2. AI-ассистент генерирует код: consumer-обёртку, кастомную SSL-фабрику, BDD-шаги, интеграцию с контекстом и хуками;

3. Инженер проверяет, корректирует и запускает — итерация за итерацией.

Это не «AI написал код, а я нажал кнопку». Это партнёрская модель: инженер отвечает за архитектуру, решения и качество, AI-агент — за скорость генерации кода, исследование альтернатив и рутинную реализацию. Результат — неделя работы за один день, при этом качество не ниже, а в некоторых аспектах выше (например, SslEngineFactory с первого раза — без багов и переделок).

Инженер + AI-агент: модель взаимодействия

Коннектор был реализован при участии AI-ассистента по архитектурным требованиям и дизайн-решениям, подготовленным инженером по автоматизации. Но важно понимать: это не разовая история про «один коннектор». Это воспроизводимая модель работы, которую можно применить к любой задаче тестовой автоматизации.

Как это работает

Почему это работает

1. Инженер не заменён — он усилен. AI-агент не принимает архитектурных решений и не несёт ответственности за качество. Но он снимает рутину и ускоряет итерации в разы.

2. Точки входа — не только в коде. Инженер, который ищет точки входа для автоматизации, находит их везде: в ручных проверках, в пропущенных багах, в медленных процессах. AI-агент помогает оцифровать эти точки и превратить их в работающие решения.

3. Масштабируемость модели. Kafka-коннектор — один пример. Тот же подход работает для API-клиентов, DB-коннекторов, CI/CD-пайплайнов, отчётности. Если инженер видит точку входа — AI-агент помогает реализовать решение за часы, а не недели.

Результат

Заключение

Статья подготовлена на основе реального опыта разработки Kafka-коннектора для тестовой автоматизации. Если вы видите в своих процессах точки входа, где ручной труд можно заменить автоматизацией, а автоматизацию усилить AI, — эта модель сработает.

P.S. Если остались какие-то вопросы — добро пожаловать в комментарии, отвечу на все!