如何使用testcontainers和spring-Kafka准备测试

Question

我正在尝试为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<String, String> consumerFactory;

  @Autowired private KafkaTemplate<Object, Object> kafkaTemplate;

  @MockBean private ActivateSomethingActivities activateCampaignActivities;

  private Consumer<String, String> consumer;

  private long initiallyCommittedOffset;

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

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

  @Test
  void shouldPerformSomething() {
...
  }

我遇到的问题很少：

1. 似乎spring-Kafka和它的@kafkalistener在所有可能的用@springboottest注释的测试中都是活动的，而不仅仅是在Kafka特定的测试中。这意味着发送到Kafka主题的消息可以被任意测试使用。首先，对于spring-Kafka的人来说，有一个问题：是否可以将spring-Kafka测试与测试容器一起使用？是否有可能为每个测试停止所有@kafkalistener并为特定的@springboottest测试显式启用它们
2. TestContainers附带了一个Kafka模块。这一个使用汇合的Kafkadocker图像，这是非常顽固的配置。例如，您不能设置一些代理属性，也不能告诉容器在开始后应该创建哪个主题。经过对这个模块的努力，我决定使用带有wurstmeister/kafka图像的docker-compose模块。后一种方法的问题是，当我使用命令行maven运行测试时，我会得到错误消息，告诉kafka已经在9092端口上运行。maven似乎在MVN测试期间启动了很少的JVM，因此静态字段kafkacontainer被初始化了很少次。为什么会发生？

Answer 1

1. 在类上使用@dirtiescontext在测试完成时关闭侦听器。

2号没主意。