Средний

16 мин

222

Привет, Хабр! На связи Егор Лаптев — 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=earliest
enable.auto.commit=false
max.poll.records=100
key.deserializer=StringDeserializer
value.deserializer=StringDeserializer

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

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

@After
public 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. Если остались какие-то вопросы — добро пожаловать в комментарии, отвечу на все!