Наверняка в вашем проекте используется очередь сообщений (не важно kafka, pulsar или какой-нибудь зайчик). Основной проблемой является подробное тестирование работы вашей системы. Рассмотрим варианты решения и посмотрим, что там у автора в рукаве.

Как решается сейчас

Очевидно, что наиболее простым вариантом тестирования интеграции с очередями сообщений является отсутствие тестирования, вернее перекладывание ответственности на КВА-инженеров. Возможно в вашем проекте даже кто-то подобное практикует, но серьезные разработчики так не поступают.

Серьезным сеньором, тестирующим очереди сообщений по полной, является так любимый либералами совок - баба Зина из магазина с ее коронной фразой:

Обоснование, почему нужен суперкомпьютер для тестирования интеграций

База зина дурного не скажет, поэтому поднимет кластер, сконфигурует как-нибудь и будет утверждать, что интеграция рабочая.

Однако у такого подхода есть множество проблем, а именно:

Отсутствует контроль над потоками данных, невозможно утверждать, что все сообщения были обработаны, что сами сообщения корректны;
Для тестов такой подход не идиоматичен, если для одного сообщения еще можно смириться, то в случае сложного дерева (из-за ветвлений и создания побочных сообщений) утверждать, что вы протестировали все что нужно - нельзя, даже если почему-то покрытие тестами 100%;
Конфигурация при тестировании отличается от того, что в проде, утверждать, что работать будет так же на 100% нельзя;
Вам нужен суперкомпьютер для тестирования, что бы ваша команда могла спокойно работать;
(Только бабе Зине не говорите) Тестирование занимает много больше времени - разработчик больше времени тратит впустую, пока ждет завершения тестов, раз уж он утверждает, что его время дорогое, то деньги тратяться дважды - на оборудование и на разработчика. Следовательно либо баба Зина ворует, либо заблуждается.

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

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

Решение

В проекте mireapay для работы с очередями сообщений специально разработана библиотека message-queue . Ее особенностью является тот факт, что она позволяет не только менять очереди сообщений (например Pulsar на Kafka) приложения без переписывания всего и вся, но так же эмулировать тестовую среду настолько, насколько это возможно, предоставляя дополнительные функции, которых у очередей сообщений быть не может. Не забываем, что вы все равно не поднимите свой кластер с нужной конфигурацией, партициями и прочим - иначе ваши тесты будут как золотой унитаз по стоимости их запуска.

Данная библиотека базируется на двух интерфейсах:

EventConsumer - реализация данного интерфейса позволяет слушать топик и получать сообщения по одному. Пакетного консьюмера нет, автору он не требовался - читатель может реализовать его самостоятельно;
EventProducer - для каждого топика будет создан бин с данным интерфейсом, подключаем его к вашей компоненте и отправляем сообщения в топик. Если отправка не удалась - ловите исключение.

Часто бывает необходимо разделить одно и то же событие на разные топики, для этого создана аннотация MessageQueueId, вешается на ваш консьюмер. По умолчанию у всех консьюмеров идентификатор - default, но вы можете задать свой собственный. Для продьюсеров нужно использовать полное имя бина, например defaultSimpleEventProducer, richSimpleEventProducer. Регистрация топиков (для создания генератором бинов) производится в конфигурации проекта. Рассмотрим на примере сервиса контрактов:

event:
  provider: "pulsar"
  consumer:
    - event-class: "com.lastrix.mps.node.model.event.contract.external.ExternalContractEvent"
  producer:
    - event-class: "com.lastrix.mps.node.model.event.contract.external.ExternalContractEvent"
      message-queue-id: "out"
      topic-group: "out"

Сервис слушает очередь сообщений для события ExternalContractEvent из топика по умолчанию, но пишет в топик с группой out, группа используется при генерации имени топика, в время как message-queue-id нужен для генерации бинов и разделения их. Т.к. события оказываются в разных топиках, то наш сервис не будет читать свои же сообщения и накручивать количество обработанных сообщений в секунду. Да, баба Зина, тебе не удастся всем говорить, что твой сервис обрабатывает 300к сообщений в секунду, потому что 299999 из них ты отпинываешь обратно, как обработанные.

Теперь, что бы тестировать такую интеграцию нам нужна конфигурация тестов:

event:
  provider: "test"
  consumer:
    - event-class: "com.lastrix.mps.node.model.event.contract.external.ExternalContractEvent"
  producer:
    - event-class: "com.lastrix.mps.node.model.event.contract.external.ExternalContractEvent"
      message-queue-id: "out"

Выглядит почти идентично, разница только в провайдере. У читателя возможно возникнет желание высказаться, что надо бы в аннотации все вынести. Не спешите. В этом же сервисе используется другое событие, которое сервис сам отправляет и читает, в том числе свои собственные.

event:
  provider: "test"
  consumer:
    - event-class: "com.lastrix.mps.node.model.event.contract.lifecycle.ContractLifecycleEvent"
  producer:
    - event-class: "com.lastrix.mps.node.model.event.contract.lifecycle.ContractLifecycleEvent"
      # this will allow us to manually control event processing
      message-queue-id: "test"

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

Теперь мы можем приступить к написанию тестов. Для упрощения работы рекомендуется разделить код, декларации при помощи ООП:

Абстрактный класс позволит нам задекларировать топики, пример:

    // все очереди сообщений одним списком, TestMessageQueue - управляет
    // как продюсером, так и консьюмером
    @Autowired
    List<TestMessageQueue<?>> messageQueues;

    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    TestMessageQueue<ExternalContractEvent> defaultExternalContractEventMessageQueue;
    // обратите внимание на префикс, он определяется идентификатором, а не группой
    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    TestMessageQueue<ExternalContractEvent> outExternalContractEventMessageQueue;

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

Рассмотрим один из методов:

    protected void assertEventStatus(Contract contract, ContractStatus status, NodeId expectedNodeId) {
        var event = testContractLifecycleEventMessageQueue.poll();
        assertNotNull(event);
        assertTrue(event instanceof FactCreatedContractLifecycleEvent);
        var factEvent = (FactCreatedContractLifecycleEvent) event;
        assertEquals(contract.getId(), factEvent.getContractId());
        var fact = defaultContractFactMapper.toModel(factEvent.getContractFact());
        assertEquals(expectedNodeId, fact.getNodeId());
        assertEquals(StatusContractFactDetails.TYPE, fact.getDetails().getType());
        var details = (StatusContractFactDetails) fact.getDetails();
        assertEquals(status, details.getStatus());

        assertAck(defaultContractLifecycleEventMessageQueue, event);
    }

Данный метод читает из нашего специального топика для тестирования, куда сервис будет на время тестирования писать сообщения. Осуществляется валидация данных и, в случае успеха, перекладывание в основной топик, что бы сервис смог прочесть его и обработать. По такому принципу строятся все эти методы с небольшими вариациями.

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

    @Test
    void test() {
        var nonLocalWalletId = Instancio.create(QWalletId.class);
        var contract = createContract(localWalletId, contractExternalId, generatePaymentTemplates(nonLocalWalletId));


        pushContractCreatedEvent(contract);
        assertSlaveReceiveContractCreated(contract);

        assertCurrentStatus(contract.getId(), ContractStatus.INIT);

        responseSlaveInitialized(contract, nonLocalWalletId.nodeId(), List.of(
                statusFact(contract.getId(), nonLocalWalletId.nodeId(), ContractStatus.INIT),
                statusFact(contract.getId(), nonLocalWalletId.nodeId(), ContractStatus.ACTIVE)
        ));

        assertEventStatus(contract, ContractStatus.ACTIVE);
        assertEventLocalStatus(contract, ContractStatus.ACTIVE);
        assertExternalEventStatusSent(contract, ContractStatus.ACTIVE);

        assertCreatedHistoryEvent(contract);

//        assertApprovalRequested(contract, localWalletId, getPaymentTemplates(contract));

//        approve(contract, localWalletId, ApprovalType.WITHDRAW);
        assertAutoApproved(contract, localWalletId, ApprovalType.WITHDRAW);

        var fromWithdrawTransactionEvent = expectCreateTransactionEvent(contract, localWalletId, ApprovalType.WITHDRAW);

        respondCreateTransactionSuccess(fromWithdrawTransactionEvent);

        assertPaymentHoldEvent(contract);
        var details = assertExternalPaymentHoldEvent(contract);

        respondExternalPaymentHold(contract, nonLocalWalletId, buildDepositDetails(nonLocalWalletId, details));
        assertPaymentHoldEvent(contract);

        respondExternalEvent(contract, statusFact(contract.getId(), nonLocalWalletId.nodeId(), ContractStatus.COMPLETING));
        assertEventStatus(contract, ContractStatus.COMPLETING, nonLocalWalletId.nodeId());
        assertEventStatus(contract, ContractStatus.COMPLETING);
        assertExternalEventStatusSent(contract, ContractStatus.COMPLETING);
        assertEventLocalStatus(contract, ContractStatus.COMPLETING);

        assertTransactionConfirmed(fromWithdrawTransactionEvent.getPaymentInfo());

        respondExternalEvent(contract, statusFact(contract.getId(), nonLocalWalletId.nodeId(), ContractStatus.COMPLETED));
        assertEventStatus(contract, ContractStatus.COMPLETED, nonLocalWalletId.nodeId());

        assertEventStatus(contract, ContractStatus.COMPLETED);
        assertExternalEventStatusSent(contract, ContractStatus.COMPLETED);
        assertEventLocalStatus(contract, ContractStatus.COMPLETED);

        assertContractStatusNotification(contract, localWalletId, ContractStatus.COMPLETED);

        assertCompleteHistoryEvent(contract);

        assertMessageQueuesAreEmpty();
    }

Данный тест проверяет успешный перевод с кошелька на кошелек, когда в обработке учавствуют два узла. Подобные тесты можно писать пачками и времени они почти не требуют на реализацию. Если бы автор писал данный тест как баба Зина, то он бы занял более 2 000 строк!!!

Заключение

В данной работе предложен вариант тестирования интеграций с очередями сообщений на основе специальной библиотеки и сценарного тестирования.

Преимуществом сценарного теста является тот факт, что даже если в результате изменений сломается сразу много сценариев, то отлаживать работу можно по очереди. Починив один - скорее всего почините сразу много. Сценарный тест - это почти на 100% бизнес логика приложения, которую при желании и начальных навыках программирования, в состоянии читать аналитик.

Побочным преимуществом такого подхода является колоссальное сокращение времени на рефакторинг кода. Какие бы изменения вы не вносили в ваш код - ваш сценарий останется неизменным, т.к. он почти не пересекается с вашим кодом. Правки в методы проверки минимальны и легко проверяемы.

При прохождении ревью кода можно сразу увидеть логику работы сценария и логику тестовых методов, появляется возможность делегировать разработку сценария и методов валидации (например пишется сценарий и загрушки методов с описанием того, что надо сделать, а какой-нибудь джун учится разработке на этих методах). Разобрать их отдельно и обсуждать тест на разных уровнях. Сами методы, разработанные для одного сценария, будут почти гарантированно использоваться в других, что поможет в дальнейшем сократить время на разработку тестов. При большой вариантивности (сложный граф состояний системы) такой подход является незаменимым.

Тестируем интеграцию с очередями сообщений правильно