Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use configure method of org.testcontainers.containers.KafkaContainer class

Best Testcontainers-java code snippet using org.testcontainers.containers.KafkaContainer.configure

Source:TestcontainersConfiguration.java

...46    @Override47    public boolean isAvailable() {48        return true;49    }50    void configureContainerNetworks(Set<GenericContainer<?>> containers, Class<?> clazz) {51        // Put all containers in the same network if no networks are explicitly defined52        boolean networksDefined = false;53        for (GenericContainer<?> c : containers) {54            networksDefined |= c.getNetwork() != null;55        }56        if (!networksDefined) {57            LOG.debug("No networks explicitly defined. Using shared network for all containers in " + clazz);58            containers.forEach(c -> c.setNetwork(Network.SHARED));59        }60    }61    @Override62    public void preConfigure(Class<?> testClass) {63        containers = discoveredContainers.computeIfAbsent(testClass, clazz -> new ContainerGroup(clazz));64        // Put all containers in the same network if no networks are explicitly defined65        if (containers.hasSharedConfig()) {66            configureContainerNetworks(containers.sharedContainers, containers.sharedConfigClass);67        }68        configureContainerNetworks(containers.unsharedContainers, testClass);69        // Give ServerAdapters a chance to do some auto-wiring between containers70        ApplicationContainer app = containers.app;71        if (app != null) {72            app.getServerAdapter().configure(containers.allContainers);73            if (isJwtNeeded() &&74                !app.isRunning() &&75                !app.getEnvMap().containsKey(JwtBuilder.MP_JWT_PUBLIC_KEY) &&76                !app.getEnvMap().containsKey(JwtBuilder.MP_JWT_ISSUER)) {77                app.withEnv(JwtBuilder.MP_JWT_PUBLIC_KEY, JwtBuilder.getPublicKey());78                app.withEnv(JwtBuilder.MP_JWT_ISSUER, JwtConfig.DEFAULT_ISSUER);79                LOG.debug("Using default generated JWT settings for " + app);80            }81        }82    }83    @Override84    public void start() {85        List<GenericContainer<?>> containersToStart = new ArrayList<>();86        long start = System.currentTimeMillis();87        // Start shared containers first88        if (containers.hasSharedConfig()) {89            try {90                SharedContainerConfiguration config = containers.sharedConfigClass.newInstance();91                config.startContainers();92                LOG.debug("Shared contianer config for " + containers.sharedConfigClass + " implemented a manual start procedure.");93            } catch (InstantiationException | IllegalAccessException e) {94                throw new ExtensionConfigurationException("Unable to instantiate " + containers.sharedConfigClass, e);95            } catch (UnsupportedOperationException ignore) {96                // This just means manual container start is not being used97                containersToStart.addAll(containers.sharedContainers);98            }99        }100        containersToStart.addAll(containers.unsharedContainers);101        containersToStart.removeIf(c -> c.isRunning());102        if (containersToStart.size() > 0) {103            LOG.info("Starting " + containersToStart.size() + " container(s) in parallel for " + containers.testClass);104            for (GenericContainer<?> c : containersToStart)105                LOG.info("  " + c.getDockerImageName());106            Startables.deepStart(containersToStart).join();107        }108        LOG.info("All containers started in " + (System.currentTimeMillis() - start) + "ms");109        configureKafka();110    }111    void configureKafka() {112        // If a KafkaContainer is defined, store the bootstrap location113        Class<?> KafkaContainer = tryLoad("org.testcontainers.containers.KafkaContainer");114        if (KafkaContainer == null)115            return;116        Set<GenericContainer<?>> kafkaContainers = containers.allContainers.stream()117                        .filter(c -> KafkaContainer.isAssignableFrom(c.getClass()))118                        .collect(Collectors.toSet());119        if (kafkaContainers.size() == 1) {120            try {121                GenericContainer<?> kafka = kafkaContainers.iterator().next();122                String bootstrapServers = (String) KafkaContainer.getMethod("getBootstrapServers").invoke(kafka);123                System.setProperty("org.microshed.kafka.bootstrap.servers", bootstrapServers);124                LOG.debug("Discovered KafkaContainer with bootstrap.servers=" + bootstrapServers);125            } catch (Exception e) {126                LOG.warn("Unable to set kafka boostrap server", e);127            }128        } else if (kafkaContainers.size() > 1) {129            LOG.info("Located multiple KafkaContainer instances. Unable to auto configure kafka clients");130        } else {131            LOG.debug("No KafkaContainer instances found in configuration");132        }133    }134    @Override135    public String getApplicationURL() {136        ApplicationContainer mpApp = containers.app;137        if (mpApp == null) {138            String sharedConfigMsg = containers.hasSharedConfig() ? " or " + containers.sharedConfigClass : "";139            throw new ExtensionConfigurationException("No public static ApplicationContainer fields annotated with @Container were located " +140                                                      "on " + containers.testClass + sharedConfigMsg + ".");141        }142        return mpApp.getApplicationURL();143    }...

Source:AbstractKsqlDbServerContainer.java

...30                //KafkaContainer "internal" port is hardcoded to 909231                .withEnv("KSQL_BOOTSTRAP_SERVERS", "PLAINTEXT://" + kafkaContainer.getNetworkAliases().get(0) + ":9092");32    }33    /**34     * Use to configure an external kafka35     * @param bootstrapServers36     * @return37     */38    public T withKafka(String bootstrapServers) {39        return withEnv("KSQL_BOOTSTRAP_SERVERS", bootstrapServers);40    }41    /**42     *43     * @param schemaRegistryContainer44     * @return45     */46    public T withSchemaRegistryContainer(SchemaRegistryContainer schemaRegistryContainer) {47        return dependsOn(schemaRegistryContainer)48                .withEnv("KSQL_KSQL_SCHEMA_REGISTRY_URL", schemaRegistryContainer.getSchemaRegistryInternalNetworkUrl());49    }50    /**51     * Use to configure an "external" schemaRegistry52     *53     * @param schemaRegistryUrl54     * @return55     */56    public T withSchemaRegistry(String schemaRegistryUrl) {57        return withEnv("KSQL_KSQL_SCHEMA_REGISTRY_URL", schemaRegistryUrl);58    }59    public String getKsqlDbUrl() {60        return "http://" + getContainerIpAddress() + ":" + getMappedPort(DEFAULT_PORT);61    }62}...

Source:AppContainerConfig.java

...37        if (ApplicationEnvironment.Resolver.isSelected(HollowTestcontainersConfiguration.class)) {38            // Run in dev mode. 39            // The application talks to KafkaContainer from outside of the Docker network,40            // and it can talk to kafka directly on 9093. 41            // The MicroProfile configure should define as following:42            // mp.messaging.connector.liberty-kafka.bootstrap.servers=localhost:909343        } else {44            // Run by maven verify goal.45            // The application talks to KafkaContainer within Docker network, 46            // and it need to talk to the broker on port 909247            kafka.withNetworkAliases("kafka");48            app.withEnv("MP_MESSAGING_CONNECTOR_LIBERTY_KAFKA_BOOTSTRAP_SERVERS", "kafka:9092");49        }50        kafka.start();51        app.start();52    }53    54}...

configure

Using AI Code Generation

1package org.testcontainers.containers;2import java.util.Properties;3import org.apache.kafka.clients.admin.AdminClient;4import org.apache.kafka.clients.admin.AdminClientConfig;5import org.apache.kafka.clients.admin.NewTopic;6import org.junit.Test;7public class KafkaContainerTest {8    public void test() throws Exception {9        try (KafkaContainer kafkaContainer = new KafkaContainer()) {10            kafkaContainer.start();11            Properties properties = new Properties();12            properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());13            AdminClient adminClient = AdminClient.create(properties);14            NewTopic newTopic = new NewTopic("test", 1, (short) 1);15            adminClient.createTopics(Collections.singletonList(newTopic));16            adminClient.close();17        }18    }19}20package org.testcontainers.containers;21import java.util.Properties;22import org.apache.kafka.clients.admin.AdminClient;23import org.apache.kafka.clients.admin.AdminClientConfig;24import org.apache.kafka.clients.admin.NewTopic;25import org.junit.Test;26public class KafkaContainerTest {27    public void test() throws Exception {28        try (KafkaContainer kafkaContainer = new KafkaContainer()29                .withConfiguration(KafkaContainer.newKafkaConfiguration())) {30            kafkaContainer.start();31            Properties properties = new Properties();32            properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());33            AdminClient adminClient = AdminClient.create(properties);34            NewTopic newTopic = new NewTopic("test", 1, (short) 1);35            adminClient.createTopics(Collections.singletonList(newTopic));36            adminClient.close();37        }38    }39}40package org.testcontainers.containers;41import java.util.Properties;42import org.apache.kafka.clients.admin.AdminClient;43import org.apache.kafka.clients.admin.AdminClientConfig;44import org.apache.kafka.clients.admin.NewTopic;45import org.junit.Test;46public class KafkaContainerTest {47    public void test() throws Exception {48        try (KafkaContainer kafkaContainer = new KafkaContainer()49                .withConfiguration(KafkaContainer.newKafkaConfiguration()50                        .with("auto.create.topics.enable", "true"))) {51            kafkaContainer.start();

configure

Using AI Code Generation

1import org.testcontainers.containers.KafkaContainer;2public class KafkaContainerTest {3    public static void main(String[] args) {4        KafkaContainer kafka = new KafkaContainer("5.2.1");5        kafka.configure();6        kafka.start();7        System.out.println(kafka.getBootstrapServers());8    }9}

configure

Using AI Code Generation

1package org.testcontainers.containers;2import java.util.HashMap;3import java.util.Map;4import org.junit.Test;5public class KafkaContainerTest {6    public void testKafkaContainer() {7        Map<String, String> envVars = new HashMap<>();8        envVars.put("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181");9        envVars.put("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");10        KafkaContainer kafkaContainer = new KafkaContainer()11            .withEnv(envVars);12        kafkaContainer.start();13    }14}15package org.testcontainers.containers;16import java.util.HashMap;17import java.util.Map;18import org.junit.Test;19public class KafkaContainerTest {20    public void testKafkaContainer() {21        Map<String, String> envVars = new HashMap<>();22        envVars.put("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181");23        envVars.put("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");24        KafkaContainer kafkaContainer = new KafkaContainer()25            .withEnv(envVars);26        kafkaContainer.start();27    }28}29package org.testcontainers.containers;30import java.util.HashMap;31import java.util.Map;32import org.junit.Test;33public class KafkaContainerTest {34    public void testKafkaContainer() {35        Map<String, String> envVars = new HashMap<>();36        envVars.put("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181");37        envVars.put("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");38        KafkaContainer kafkaContainer = new KafkaContainer()39            .withEnv(envVars);40        kafkaContainer.start();41    }42}

configure

Using AI Code Generation

1package org.testcontainers.containers;2import org.junit.Test;3import java.util.Map;4public class KafkaContainerTest {5    public void testConfigure() {6        KafkaContainer kafkaContainer = new KafkaContainer();7        kafkaContainer.configure(Map.of("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1"));8        kafkaContainer.start();9        kafkaContainer.stop();10    }11}12package org.testcontainers.containers;13import org.junit.Test;14import java.util.Map;15public class KafkaContainerTest {16    public void testConfigure() {17        KafkaContainer kafkaContainer = new KafkaContainer();18        kafkaContainer.configure(Map.of("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1"));19        kafkaContainer.start();20        kafkaContainer.stop();21    }22}23package org.testcontainers.containers;24import org.junit.Test;25import java.util.Map;26public class KafkaContainerTest {27    public void testConfigure() {28        KafkaContainer kafkaContainer = new KafkaContainer();29        kafkaContainer.configure(Map.of("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1"));30        kafkaContainer.start();31        kafkaContainer.stop();32    }33}34package org.testcontainers.containers;35import org.junit.Test;36import java.util.Map;37public class KafkaContainerTest {38    public void testConfigure() {39        KafkaContainer kafkaContainer = new KafkaContainer();40        kafkaContainer.configure(Map.of("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1"));41        kafkaContainer.start();42        kafkaContainer.stop();43    }44}45package org.testcontainers.containers;46import org.junit.Test;47import java.util.Map;48public class KafkaContainerTest {49    public void testConfigure() {50        KafkaContainer kafkaContainer = new KafkaContainer();51        kafkaContainer.configure(Map.of("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1"));52        kafkaContainer.start();53        kafkaContainer.stop();54    }55}

configure

Using AI Code Generation

1public class 1 {2  public static void main(String[] args) {3    KafkaContainer kafka = new KafkaContainer();4    kafka.withNetwork(Network.newNetwork());5    kafka.withNetworkAliases("kafka");6    kafka.withExposedPorts(9092);7    kafka.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");8    kafka.withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true");9    kafka.withEnv("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181");10    kafka.start();11    System.out.println(kafka.getBootstrapServers());12  }13}14public class 2 {15  public static void main(String[] args) {16    KafkaContainer kafka = new KafkaContainer();17    kafka.withNetwork(Network.newNetwork());18    kafka.withNetworkAliases("kafka");19    kafka.withExposedPorts(9092);20    kafka.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");21    kafka.withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true");22    kafka.withEnv("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181");23    kafka.start();24    System.out.println(kafka.getBootstrapServers());25  }26}27public class 3 {28  public static void main(String[] args) {29    KafkaContainer kafka = new KafkaContainer();30    kafka.withNetwork(Network.newNetwork());31    kafka.withNetworkAliases("kafka");32    kafka.withExposedPorts(9092);33    kafka.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");34    kafka.withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true");35    kafka.withEnv("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181");36    kafka.start();37    System.out.println(kafka.getBootstrapServers());38  }39}

configure

Using AI Code Generation

1import org.testcontainers.containers.KafkaContainer;2public class KafkaContainerEx {3    public static void main(String[] args) {4        KafkaContainer kafka = new KafkaContainer();5        kafka.withEmbeddedZookeeper();6        kafka.withNetwork(null);7        kafka.withNetworkAliases(null);8        kafka.withCreateContainerCmdModifier(null);9        kafka.withExposedPorts(9092, 29092);10        kafka.withEnv(null);11        kafka.withEnv(null, null);12        kafka.withEnv(null, null, null);13        kafka.withEnv(null, null, null, null);14        kafka.withEnv(null, null, null, null, null);15        kafka.withEnv(null, null, null, null, null, null);16        kafka.withEnv(null, null, null, null, null, null, null);17        kafka.withEnv(null, null, null, null, null, null, null, null);18        kafka.withEnv(null, null, null, null, null, null, null, null, null);19        kafka.withEnv(null, null, null, null, null, null, null, null, null, null);20        kafka.withEnv(null, null, null, null, null, null, null, null, null, null, null);21        kafka.withEnv(null, null, null, null, null, null, null, null, null, null, null, null);22        kafka.withEnv(null, null, null, null, null, null, null, null, null, null, null, null, null);23        kafka.withEnv(null, null, null, null, null, null, null, null, null, null, null, null, null, null);24        kafka.withEnv(null, null, null, null, null, null, null, null, null, null, null, null, null, null, null);25        kafka.withEnv(null, null, null, null, null, null, nul

configure

Using AI Code Generation

1package org.testcontainers.containers;2import org.junit.Test;3import org.testcontainers.containers.KafkaContainer;4import org.testcontainers.utility.DockerImageName;5public class KafkaContainerTest {6    public void testKafka() {7        KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.0"));8        kafka.withNetworkAliases("kafka");9        kafka.configure("2.3.1");10        kafka.start();11    }12}13package org.testcontainers.containers;14import org.junit.Test;15import org.testcontainers.containers.KafkaContainer;16import org.testcontainers.utility.DockerImageName;17public class KafkaContainerTest {18    public void testKafka() {19        KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.0"));20        kafka.withNetworkAliases("kafka");21        kafka.configure("2.3.1");22        kafka.start();23    }24}25package org.testcontainers.containers;26import org.junit.Test;27import org.testcontainers.containers.KafkaContainer;28import org.testcontainers.utility.DockerImageName;29public class KafkaContainerTest {30    public void testKafka() {31        KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.0"));32        kafka.withNetworkAliases("kafka");33        kafka.configure("2.3.1");34        kafka.start();35    }36}37package org.testcontainers.containers;38import org.junit.Test;39import org.testcontainers.containers.KafkaContainer;40import org.testcontainers.utility.DockerImageName;

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.