Как протестировать логику консьюмеров и продюсеров и не сгореть? Spring Boot 3, Spring Kafka
Всем привет. Решил поделиться опытом тестирования логики консьюмеров и продюсеров в стандартном Spring Boot приложении. Я несколько раз подходил к этой задаче с различными вариантами и находил разные подводные камни, которые стимулировали меня искать дальше - более лучшее решение. И вот, в очередной раз прошерстив накопленный опыт человечества (stackoverflow), я реализовал очередной вариант, в котором пока не нашел минусов предыдущих реализаций. В любом случае, поделюсь с вами как я к этому пришел и почему для моих кейсов не сработали другие варианты (или показались хуже последнего).
Дисклеймер. Я сторонник интеграционных тестов при запуске сборки проекта с тестами. То есть это когда для unit-тестов поднимается контекст Spring со всеми плюсами (окружение, максимально похожее на боевое) и минусами (это ДОЛГО запускается и ДОЛГО описывается).
Часть 1. А разве это проблема вообще?
Коротко: да.
Началось все с того, чтобы признать, что логику внутри консьюмера и продюсера нужно тестировать. Как выглядит самый стандартный консьюмер? Ну, предположу, что как-то так:
@Slf4j
@Service
@RequiredArgsConstructor
public class UserConsumer {
private final UserService userService;
private final NotUserService notUserService;
private final UserHandler userHandler;
private final UserMapper userMapper;
private final List<String> userCodes;
@KafkaListener(groupId = "userConsumerGroupId",
clientIdPrefix = "UserConsumer",
topics = {"user-topic"},
containerFactory = "kafkaListenerContainerFactory")
public void consume(UserInfoRecord userInfoRecord) {
log.info("Тут разнообразная логика обработки");
log.info("Cохранение в бд?");
log.info("Работа с внешними сервисами?");
log.info("Вызов мапперов?");
log.info("Отправка информации в Kafka?");
}
}
В методе consume() как раз может скрываться нетривиальная логика, а может быть даже тривиальная, но которую хочется проверить до того, как отдавать заказчику.
А как выглядит стандартный продюсер? Ну, наверное, как-то так:
@Slf4j
@Service
@RequiredArgsConstructor
public class UserProducer {
private final UserService userService;
/** Если у нас какой-то персистинг сообщений перед отправкой или обработка ошибок отправки */
private final KafkaMessageService kafkaMessageService;
/** Если мы просто отправляем - используем KafkaProducer, например, у нас по бОльшей части используетс AVRO */
private final KafkaProducer<String, SpecificRecordBase> kafkaProducer;
/** Но есть места, где используется JSON вместо AVRO */
private final KafkaProducer<String, String> kafkaJsonProducer;
private final ApplicationProperties applicationProperties;
public void sendUserInfoMessage(User user, Set<Integer> managerIds) {
log.info("Тут логика отправки");
log.info("Обычно сначала идет сборка модели для отправки.")
log.info("Но может быть и какая-то более сложная логика.");
var record = UserInfoRecord.newBuilder().build();
log.info("Тут может быть как непосредственно отправка в Kafka");
log.info("Так и реализация outbox-паттерна.");
kafkaMessageService.persistMessageToSend(applicationProperties.userTopic(), record);
}
}Может быть используется outbox-pattern, и тогда мы перед отправкой просто кладем в бд сообщение, которое надо отправить.Или кладем в БД только те сообщения, которые не удалось по каким-то причинам отправить. Или не в БД кладем, а еще куда отправляем. Сути это не меняет - мы собираем сообщение и отправляем его либо в бд, либо сразу в Kafka с помощью KafkaProducer. Мы не хотим проверять в каждом интеграционном тесте логику непосредственной отправки сообщения, логику персиста сообщений - это мы делаем в стартере работы с Kafka. А в конечных сервисах считаем, что все протестировано и все работает корректно. Но очень хочется проверить логику ДО, то есть как собирается модель, корректно ли мы кладем поля, не забыли ли мы докинуть в модель недостающие данные, правильно ли мы вызываем калькулятор роста пользователя на основании количества букв в его ФИО.
Итого: логика в консьюмерах и продюсерах тоже просится в unit-тесты, как и остальной код.
Часть 1.5. Промежуточные варианты
Например, использовать тестовый стенд Kafka, то есть подключаться к нему прямо из тестов. Мне кажется, это плохой вариант, когда у тебя появляется зависимость на внешний сервис для выполнения тестов.
Сюда же можно отнести вариант поднятия Kafka (хоть в контейнере, например) прямо на машине-сборщике (Gitlab Runner в нашем случае). Уже лучше, чем абзацем выше, но выглядит такая конструкция слабоподдерживаемой.
Часть 2. "Просто проверь логику, забей на логи"
Да, первый этап - просто проверить логику, несмотря на то, что в консоль при поднятии контекста Spring постоянно пишутся ошибки подключения к Kafka:
[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] requestId: INFO o.apache.kafka.clients.NetworkClient - [Consumer clientId=Consumer-0, groupId=app-group] Node -1 disconnected.
[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] requestId: WARN o.apache.kafka.clients.NetworkClient - [Consumer clientId=Consumer-0, groupId=app-group] Connection to node -1 (localhost/127.0.0.1:3333) could not be established. Broker may not be available.
[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] requestId: WARN o.apache.kafka.clients.NetworkClient - [Consumer clientId=Consumer-0, groupId=app-group] Bootstrap broker localhost:3333 (id: -1 rack: null) disconnectedВозможно, это вполне рабочий вариант и эти ошибки коннекта никак не помешают работе всех тестов при сборке. Но если консьюмеров в одном приложении 5-10, то таких логов становится ощутимо много (например, может быть больше половины общего объема логов при тестах), и в целом такое решение может влиять на работоспособность тестов.
Часть 3. "Да замокай ты да и всё.."
Второй совет, который я услышал или увидел - просто замокать консьюмеры и/или продюсеры. То есть вот так (допустим, для интеграционных тестов мы используем некий BaseTest для конфигурирования контекста):
@SpringJUnitConfig
@SpringBootTest
@EnableConfigurationProperties(ApplicationProperties.class)
public abstract class BaseTest {
@MockBean(name = "kafkaProducer")
protected KafkaProducer kafkaProducer;
@MockBean
protected UserConsumer userConsumer;
}Это определенно поможет не ловить ошибки коннекта к Kafka, потому что ни консьюмеры, ни продюсеры в таком случае не будут конфигурироваться для контекста. А просто будут замоканы.
И кстати, для продюсера это вполне оптимальный вариант! Мокая KafkaProducer - то есть бин отправки сообщений в Kafka, мы оставляем бин класса-продюсера в контексте и можем легко написать на него тест:
class UserProducerTest extends BaseTest {
@Autowired
private UserProducer userProducer;
@Test
void testUserProducer() {
log.info("тестируем логику");
userProducer.produce();
}
}Как в таком случае написать тесты на консюмер? Ну, например, так:
class UserConsumerTest extends BaseTest {
private final UserConsumer userConsumer = new UserConsumer(<все зависимости>);
@Test
void testUserProducer() {
log.info("тестируем логику");
userConsumer.consume();
}
}Помните, что у нас в UserConsumer может быть много зависимостей на другие сервисы/классы? Их все нужно либо подготовить тут, либо взять из контекста и подготовить только консьюмер. В любом случае это получается ручная подготовка класса.
Итого, замокать продюсер - ок, замокать консьюмер? Сомнительно, но ОКЭ - придется инициализировать консьюмеры вручную (а при запуске приложения это делает Spring, получается тесты происходят в контексте, но условия запуска разные).
Часть 4. "Да просто возьми EmbeddedKafka, она же для этого и сделана!"
Да, есть такой вариант - поднять EmbeddedKafka для тестов. Настраивается такой вариант несложно - вешаем аннотацию на тест:
@EmbeddedKafka(partitions = 1, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" })И добавляем зависимость в maven (в моем случае):
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
</dependency>Какие плюсы? Буквально одной строчкой настройки у нас поднимается простенькая embedded Kafka в тестах, которая позволяет проверить полный флоу. Например проверить, что мы примем в консьюмере именно то, что отправили в продюсере.
Более подробно об этом и следующем варианте написано, например, тут: https://www.baeldung.com/spring-boot-kafka-testing
Я лишь скажу минус, который нашел: зависимость spring-kafka-test тянет за собой транзитивно зависимости на scala:
jackson-module-scala_2.13
scala-collection-compat_2.13
scala-java8-compat_2.13
scala-library
scala-logging_2.13
scala-reflect
И в целом-то можно было бы сказать что "ну и ладно, ну и пускай, все равно это только в тестах". Но эти либы влияют на исполнение кода! Я, например, словил такую ошибку:
ClassCast class scala.collection.immutable.$colon$colon cannot be cast to class java.util.List (scala.collection.immutable.$colon$colon is in unnamed module of loader 'app'; java.util.List is in module java.base of loader 'bootstrap')В коде мы конечно же не использовали данный класс и в принципе пакет scala. Но мы используем querydsl. И при кодогенерации происходит "подстава". Возможно, это легко чинится более точной настройкой maven-плагина, но копаться там не захотелось.
Часть 5. "Ну тогда тебе поможет testcontainers, запускай на нем Kafka!"
И правда, поможет. У последних версий Spring Boot хорошая интеграция с testcontainers, что позволяет очень просто настраивать поднятие контейнеров в тестах.
Ах да, в начале я вскользь упомянул, что мы используем AVRO для описания структуры сообщений, передаваемых по Kafka. То есть мы используем Schema Registry (SR). А значит в тестах нужно решить проблему взаимодействия с SR. Тут тоже есть несколько вариантов - замокать через WireMock, использовать SR тестового стенда, поднимать отдельную на Gitlab Runner или еще где-то. Я попробовал два варианта - эмулировал настоящую SR через wiremock - громоздко, приходится докидывать эмуляции новых схем, обновлять текущие. В какой-то момент мы просто настроили взаимодействие с централизованной тестовой SR компании. Но тут сложность в том, что тебе обязательно нужно зарегистрировать схему в SR до запуска тестов. А тесты вообще говоря прогоняются ДО релиза проектов на тестовый стенд.
Чтобы было удобнее подключать kafka-testcontainers в разных сервисах, я создал отдельную либу для теста, с такими зависимостями:
<dependencies>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<scope>compile</scope>
</dependency>
<!-- Тут у нас просто либа с модельками и зависимостью на AVRO -->
<dependency>
<groupId>ru.alfastrah</groupId>
<artifactId>avro-models</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.wiremock</groupId>
<artifactId>wiremock-standalone</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-testcontainers</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<scope>test</scope>
</dependency>
</dependencies>Ну а основной класс в либе такой:
@ImportTestcontainers
public interface KafkaBaseTest {
Logger log = org.slf4j.LoggerFactory.getLogger(KafkaTestingClass.class);
/** Kafka-контейнер */
KafkaContainer KAFKA_CONTAINER = new KafkaContainer(KAFKA_IMAGE_NAME);
/** Динамическая конфигурация свойств подключения к БД */
@DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
if (!KAFKA_CONTAINER.isRunning()) {
KAFKA_CONTAINER.start();
var servers = KAFKA_CONTAINER.getBootstrapServers();
log.info("Kafka-контейнер запущен по адресу {} (и адрес положен в свойство bootstrap-servers)", servers);
registry.add("bootstrap-servers", () -> servers);
}
}
}Все это позволяет включить в конечном сервисе kafka на testcontainers так (implements KafkaBaseTest):
@SpringJUnitConfig
@SpringBootTest
@EnableConfigurationProperties(value = ApplicationProperties.class)
public abstract class BaseTest implements KafkaBaseTest {
}Минусы:
сборка стала дольше, так как теперь запускается контейнер с kafka;
мы проверяем не совсем то, что хотим изначально - не логику консьюмера или продюсера, а конфигурацию взаимодействия с kafka. Это, конечно, тоже важно, но;
код в тестах стал довольно громоздким, например:
class UserProducerTest extends BaseTest {
@Autowired
private KafkaSettings kafkaSettings;
@Autowired
private ApplicationProperties appProperties;
@Autowired
private UserProducer userProducer;
@Autowired
private KafkaListenerContainerFactory kafkaListenerContainerFactory;
@Autowired
private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
@Test
void KafkaProducerTest() {
var counter = new AtomicInteger(0);
new KafkaListenerCreator<SomeEvent>(kafkaSettings.groupId(), kafkaListenerContainerFactory, kafkaListenerEndpointRegistry)
.createAndRegisterListener(appProperties.topicName(), "KafkaProducerServiceTestListener", (record) -> {
log.info("Получено сообщение: {}", record);
if ("UserProducerTest".equals(record.getTestData())) {
counter.getAndIncrement();
}
});
AtomicInteger iteration = new AtomicInteger(0);
Arrays.stream(EventType.values()).forEach(eventType -> {
var record = createNewRecord(iteration.getAndIncrement());
userProducer.send(record);
});
await().atMost(Duration.ofSeconds(6L)).untilAsserted(() -> {
assertEquals(EventType.values().length, iteration.get());
assertEquals(2, counter.get());
});
}
}Часть 6. Если не всё, что выше, то что же?
Получая некоторые сложности от варианта выше, вопрос с тестированием логики был все еще не закрыт. Я окунулся в очередной раз в накопленный опыт человечества и нашел там занимательную штуку, а именно: параметр autoStartup в аннотации @KafkaListener (сразу сюда я почему-то не посмотрел).
И тогда мы забываем почти все, что написано выше и пишем следующую конструкцию:
// В тестах мокаем бины продюсеров сообщений в Kafka
public abstract class BaseTest {
@MockBean(name = "kafkaProducer")
protected KafkaProducer kafkaProducer;
@MockBean(name = "kafkaJsonProducer")
protected KafkaProducer kafkaJsonProducer;
}@Slf4j
@Service
@RequiredArgsConstructor
public class UserConsumer {
private final UserService userService;
private final NotUserService notUserService;
private final UserHandler userHandler;
private final UserMapper userMapper;
private final List<String> userCodes;
@KafkaListener(
// прописываем по умолчанию и возможность повлиять на него через свойство
autoStartup = "${auto-startup:true}",
groupId = "userConsumerGroupId",
clientIdPrefix = "UserConsumer",
topics = {"user-topic"},
containerFactory = "kafkaListenerContainerFactory")
public void consume(UserInfoRecord userInfoRecord) {
log.info("Тут разнообразная логика обработки");
log.info("Cохранение в бд?");
log.info("Работа с внешними сервисами?");
log.info("Вызов мапперов?");
log.info("Отправка информации в Kafka?");
}
}# в тестовом application.yml выключаем автозапуск:
auto-startup: falseЧто получаем в итоге: бин консьюмера в контексте, при инициализации консьюмер не пытается подключиться к kafka. При этом у нас работает валидация на уровне "неправильно задал containerFactory" - свалится ошибка при запуске или при тестах.
И тогда тест на консьюмер будет выглядеть так:
class UserConsumerTest extends BaseTest {
// Просто берем бин из контекста и проверяем метод consume
@Autowired
private UserConsumer userConsumer;
@Test
@DataSet(value = "userConsumer_initial.yml")
@ExpectedDataSet(value = "userConsumer_expected.yml")
void testConsume() {
userConsumer.consume(buildMessage());
}
}А вот пример проверки продюсера:
class UserProducerTest extends BaseTest {
@Autowired
private UserProducer userProducer;
@Autowired
private ApplicationProperties appProperties;
@Test
@DataSet(value = {"UserProducerTest.yml"})
void testPolicyPaymentProducer() {
getUsers().forEach(user -> userProducer.sendInfo(user));
verify(kafkaMessageService, times(4)).persistMessageToSend(eq(appProperties.topicName()), argThat((arg) -> {
var record = (UserRecord) arg;
assertEquals(expectedField1, record.getId());
assertEquals(expectedField2, record.getStatus());
assertEquals(expectedField3, record.getUid());
return record.getExpectedSomething() == null;
}));
}
}Пока я остановился на этом варианте как на наиболее простом в описании и использовании и при этом он позволяет проверить то, что мы изначально и хотим - логику внутри продюсеров и консьюмеров.
Спасибо, что дочитали. Расскажете в комментариях, какой вариант используете вы?