Как подготовить тесты с использованием тест-контейнеров и Spring-Kafka ⇐ JAVA

[Anonymous]

Я пытаюсь настроить интеграционные тесты для обмена сообщениями Kafka и переключиться с использования Embedded-Kafka на Testcontainers. Учитывая следующую конфигурацию для docker-compose и базовый класс для всех интеграционных тестов:
kafka-compose.yaml:

Код: Выделить всё

version: '3.3'

services:
zookeeper:
image: "wurstmeister/zookeeper"
kafka:
image: "wurstmeister/kafka:2.12-2.2.2"
ports:
- "9092:9092"
depends_on:
- "zookeeper"
environment:
KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
KAFKA_ADVERTISED_HOST_NAME: "${KAFKA_HOST:-localhost}"
KAFKA_ADVERTISED_PORT: "9092"
KAFKA_CREATE_TOPICS: "recoverer-test:1:1,some-topic"
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"

Код: Выделить всё

@SpringBootTest
@Slf4j
public class IntegrationTest {
private static final DockerComposeContainer kafkaContainer = initializeKafkaContainer();

protected static DockerComposeContainer initializeKafkaContainer() {
log.info(
"Initializing kafka container. Should be called only once. Current value of the kafkaContainer: {}",
kafkaContainer);
try {
var kafkaContainer =
new DockerComposeContainer(new File("src/test/resources/kafka-compose.yml"))
.withExposedService("kafka_1", 9092);
kafkaContainer.start();

var bootstrapServers =
format(
"PLAINTEXT://%s:%s",
kafkaContainer.getServiceHost("kafka_1", 9092),
kafkaContainer.getServicePort("kafka_1", 9092));

System.setProperty("spring.embedded.kafka.brokers", bootstrapServers);

return kafkaContainer;
} catch (Throwable t) {
log.error("Can't initialize the Kafka test container.", t);
throw t;
}
}

Код: Выделить всё

@DirtiesContext(classMode = ClassMode.BEFORE_CLASS)
class PerformSomethingInboundAdapterTest extends IntegrationTest {

private static final String GROUP_ID = "test-group-id";
private static final TopicPartition PARTITION = new TopicPartition(SOME_TOPIC, 0);
private static final Instant RECEIVED_AT = now();
private static final CustomerNumber CUSTOMER_NUMBER = CustomerNumber.of(600830);

@Autowired private KafkaListenerEndpointRegistry kafkaListenerRegistry;

@Autowired private ConsumerFactory consumerFactory;

@Autowired private KafkaTemplate kafkaTemplate;

@MockBean private ActivateSomethingActivities activateCampaignActivities;

private Consumer consumer;

private long initiallyCommittedOffset;

@BeforeEach
void startKafkaListener() {
kafkaListenerRegistry.getListenerContainers().forEach(Lifecycle::start);
}

@AfterEach
void stopKafkaListener() {
kafkaListenerRegistry.getListenerContainers().forEach(Lifecycle::stop);
}

@Test
void shouldPerformSomething() {
...
}

Я столкнулся с несколькими проблемами:

• Кажется, что Spring-kafka и это @KafkaListeners активны во всех возможных тестах, помеченных @SpringBootTest, а не только во время конкретных тестов Kafka. Это означает, что сообщение, отправленное в тему Kafka, может быть использовано произвольным тестом. Прежде всего вопрос к ребятам из Spring-Kafka: можно ли вообще использовать Spring-kafka-test вместе с Testcontainers? Есть ли какая-то возможность остановить все @KafkaListener для каждого теста и явно включить их для определенных тестов @SpringBootTest?
• Testcontainers поставляется с модулем Kafka на борту. Здесь используется слитный образ докера Kafka, который очень упрям ​​в настройке. Например, вы не можете установить некоторые свойства брокера и не можете указать контейнеру, какую тему следует создать после запуска. После борьбы с этим модулем я решил использовать модуль docker-compose с образом wurstmeister/kafka. Проблема с последним подходом заключается в том, что когда я запускаю тесты с помощью maven из командной строки, я получаю сообщение об ошибке, сообщающее, что kafka уже работает на порту 9092. Кажется, что maven запускает несколько JVM во время теста mvn, и, как следствие, статическое поле kafkaContainer инициализируется несколько раз. Почему это происходит?

Подробнее здесь: https://stackoverflow.com/questions/626 ... ring-kafka