How to prepare tests using testcontainers and spring-kafka

I'm trying to setup my integration tests for kafka messaging and to switch from using Embedded-Kafka in favor to the Testcontainers. Given following configuration for docker-compose and a base class for all integration tests:

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() {
...
  }

There are few problems which I encountered:

1. It seems that spring-kafka and it's @KafkaListeners are active during all possible tests annotated with @SpringBootTest and not only during the kafka specific ones. That means a message sent to a kafka topic can be consumed by an arbitrary test. First of all the question for the spring-kafka guys: is it possible to use spring-kafka-test along with Testcontainers at all? Are there some possibility to stop all the @KafkaListener's for each test and enable them explicitly for specific @SpringBootTest tests?
2. Testcontainers comes with a Kafka module on board. This one uses the confluent kafka docker image which is very stubborn in regards on configuration. For instance you can't set some broker properties and you can't tell the container which topic should be created after the start. After struggling with this module I decided to use the docker-compose module with the wurstmeister/kafka image. The problem with latter approach is that when I run the tests with command-line maven I get the error message telling that kafka is already running on 9092 port. It seems that maven starts few JVM during mvn test and as consequence the static field kafkaContainer gets initialised few times. Why does it happen?

1. Use @DirtiesContext on the class to shut down the listeners when a test completes.

No idea for #2.

1st might be overcome by using different contexts for different tests

Example:

@ExtendWith(SpringExtension.class)
@DataJpaTest
@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
@TestExecutionListeners({DependencyInjectionTestExecutionListener.class,
        FlywayTestExecutionListener.class})
@FlywayTest
@ActiveProfiles({"test"})
public abstract class AbstractDatabaseTest {

This is a test I created for integration testing, I would just extend this test where I needed the database layer, you could try creating a similar one for Kafka tests.
In general @SpringBootTest boostraps the entire application which might take longer and I don't like it personally.

2nd it's hard to say. You may try to print a stacktrace in initializeKafkaContainer() to see which test does that. Or if you try to go with the previous approach, you can do initialization in a static block in abstract class and then every test that extends it will use the existing static container and it will be initilialized only once.