Running 2 Kafka Containers together

[rajjaiswalsaumya]

I am trying to simulate a behavior in tests where I have to run 2 Kafka test containers where the message will be read from one Kafka Cluster and sent to another Kafka cluster. Somehow I am unable to run this and my test containers does not start.

Here is a class I am using to run test containers in java

import java.util.HashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.lifecycle.Startable;
import org.testcontainers.utility.DockerImageName;

@Slf4j
public class KafkaTestContainer implements Startable {

  // Pulling the image from private repo. If a new image needs to be used then make sure that is published to the private repo.
  // We cannot access public repos from build machine.
  //@formatter:off
  private static final DockerImageName imageName = DockerImageName.parse("confluentinc/cp-kafka:6.2.1")
                                                                  .asCompatibleSubstituteFor("kafka");
  //@formatter:on
  private KafkaContainer kafkaContainer;

  @Override
  public void start() {
    kafkaContainer = new KafkaContainer(imageName);
    kafkaContainer.start();
  }

  @Override
  public void stop() {
    kafkaContainer.stop();
  }

  public Map<String, Object> getConsumerKafkaConfig() {
    var map = new HashMap<String, Object>();
    map.putIfAbsent("kafka.bootstrap-servers", kafkaContainer.getBootstrapServers());
    return map;
  }

  public Map<String, Object> getProducerKafkaConfig() {
    var map = new HashMap<String, Object>();
    map.putIfAbsent("kafka-ext.bootstrap-servers", kafkaContainer.getBootstrapServers());
    return map;
  }

}

I am running this class in a test class as

import java.util.stream.Stream;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.testcontainers.lifecycle.Startable;
import org.testcontainers.lifecycle.Startables;

@ExtendWith(SpringExtension.class)
@SpringBootTest
@AutoConfigureMockMvc
@Slf4j
@PropertySource(value = {
    "classpath:/vault.properties",
    "classpath:/service-test.properties"
})
public class ServiceLauncherITTest {

  private static KafkaTestContainer kafkaTestSourceContainer;
  private static KafkaTestContainer kafkaTestTargetContainer;

  @BeforeAll
  @SneakyThrows
  static void beforeAll() {
    kafkaTestSourceContainer = new KafkaTestContainer();
    kafkaTestTargetContainer = new KafkaTestContainer();
    Startables
        .deepStart(kafkaTestSourceContainer /*, kafkaTestTargetContainer*/)
        .join();
  }

  @DynamicPropertySource
  static void registerProperties(DynamicPropertyRegistry registry) {
    kafkaTestSourceContainer
        .getConsumerKafkaConfig()
        .entrySet()
        .forEach(entry -> {
          registry.add(entry.getKey(), () -> entry.getValue());
        });

    kafkaTestTargetContainer
        .getProducerKafkaConfig()
        .entrySet()
        .forEach(entry -> {
          registry.add(entry.getKey(), () -> entry.getValue());
        });
  }


  @Test
  public void contextLoads() {

    log.info("Context loaded successfully.");
  }

  @AfterAll
  static void afterAll() {
    Stream
        .of(postgresTestContainer, kafkaTestSourceContainer)
        .forEach(Startable::stop);
  }
}

Can someone please point what is wrong here ?

[eddumelendez]

There is a example to set up a Kafka Cluster. See https://github.com/testcontainers/testcontainers-java/blob/main/examples/kafka-cluster/src/test/java/com/example/kafkacluster/KafkaContainerCluster.java