How to use start method of com.example.kafkacluster.KafkaContainerCluster class

Best Testcontainers-java code snippet using com.example.kafkacluster.KafkaContainerCluster.start

Source:KafkaContainerCluster.java

...82    /**83     * Used by test container lifecycle itself.84     */85    @Override86    public void start() {87        Stream<Startable> startables = this.brokers.stream().map(Startable.class::cast);88        try {89            Startables.deepStart(startables).get(60, SECONDS);90        } catch (InterruptedException | ExecutionException | TimeoutException e) {91            e.printStackTrace();92        }93        Unreliables.retryUntilTrue(60, TimeUnit.SECONDS, () -> {94            Container.ExecResult result = this.zookeeper.execInContainer(95                    "sh", "-c",96                    "zookeeper-shell zookeeper:" + KafkaContainer.ZOOKEEPER_PORT + " ls /brokers/ids | tail -n 1"97            );98            String brokers = result.getStdout();99            return brokers != null && brokers.split(",").length == this.brokersNum;100        });101    }102    /**103     * Used by test container lifecycle itself....

Source:KafkaContainerClusterTest.java

...26    public void testKafkaContainerCluster() throws Exception {27        try (28            KafkaContainerCluster cluster = new KafkaContainerCluster("6.2.1", 3, 2)29        ) {30            cluster.start();31            String bootstrapServers = cluster.getBootstrapServers();32            assertThat(cluster.getBrokers()).hasSize(3);33            testKafkaFunctionality(bootstrapServers, 3, 2);34        }35    }36    protected void testKafkaFunctionality(String bootstrapServers, int partitions, int rf) throws Exception {37        try (38            AdminClient adminClient = AdminClient.create(ImmutableMap.of(39                AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers40            ));41            KafkaProducer<String, String> producer = new KafkaProducer<>(42                ImmutableMap.of(43                    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,44                    ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString()...

start

Using AI Code Generation

1package com.example.kafkacluster;2import org.apache.kafka.clients.consumer.Consumer;3import org.apache.kafka.clients.consumer.ConsumerRecord;4import org.apache.kafka.clients.consumer.ConsumerRecords;5import org.apache.kafka.clients.consumer.KafkaConsumer;6import org.apache.kafka.common.serialization.StringDeserializer;7import org.slf4j.Logger;8import org.slf4j.LoggerFactory;9import java.time.Duration;10import java.util.Arrays;11import java.util.Properties;12public class KafkaConsumerCluster {13    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerCluster.class.getName());14    public static void main(String[] args) {15        Properties properties = new Properties();16        properties.setProperty("bootstrap.servers", "localhost:9092,localhost:9093");17        properties.setProperty("key.deserializer", StringDeserializer.class.getName());18        properties.setProperty("value.deserializer", StringDeserializer.class.getName());19        properties.setProperty("group.id", "my-fourth-application");20        properties.setProperty("auto.offset.reset", "earliest");21        Consumer<String, String> consumer = new KafkaConsumer<String, String>(properties);22        consumer.subscribe(Arrays.asList("first_topic"));23        while (true) {24            for (ConsumerRecord<String, String> record : records) {25                logger.info("Key: " + record.key() + ", Value: " + record.value());26                logger.info("Partition: " + record.partition() + ", Offset:" + record.offset());27            }28        }29    }30}31package com.example.kafkacluster;32import org.apache.kafka.clients.producer.*;33import org.apache.kafka.common.serialization.StringSerializer;34import org.slf4j.Logger;35import org.slf4j.LoggerFactory;36import java.util.Properties;37public class KafkaProducerCluster {38    public static void main(String[] args) {39        final Logger logger = LoggerFactory.getLogger(KafkaProducerCluster.class);40        Properties properties = new Properties();41        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093");42        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());43        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

start

Using AI Code Generation

1import com.example.kafkacluster.KafkaContainerCluster;2import org.apache.kafka.clients.producer.KafkaProducer;3import org.apache.kafka.clients.producer.Producer;4import org.apache.kafka.clients.producer.ProducerRecord;5import org.slf4j.Logger;6import org.slf4j.LoggerFactory;7import java.util.Properties;8import java.util.concurrent.ExecutionException;9public class KafkaContainerClusterExample {10    private static final Logger LOG = LoggerFactory.getLogger(KafkaContainerClusterExample.class);11    public static void main(String[] args) throws InterruptedException, ExecutionException {12        KafkaContainerCluster kafka = new KafkaContainerCluster(3);13        kafka.start();14        Properties props = new Properties();15        props.put("bootstrap.servers", kafka.getBootstrapServers());16        props.put("acks", "all");17        props.put("retries", 0);18        props.put("batch.size", 16384);19        props.put("linger.ms", 1);20        props.put("buffer.memory", 33554432);21        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");22        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");23        Producer<String, String> producer = new KafkaProducer<>(props);24        for (int i = 0; i < 100; i++)25            producer.send(new ProducerRecord<>("test", Integer.toString(i), Integer.toString(i))).get();26        producer.close();27        kafka.stop();28    }29}30package com.example.kafkacluster;31import org.testcontainers.containers.KafkaContainer;32public class KafkaContainerCluster {33    private final KafkaContainer[] kafkaContainers;34    public KafkaContainerCluster(int numberOfKafkaNodes) {35        this.kafkaContainers = new KafkaContainer[numberOfKafkaNodes];36        for (int i = 0; i < numberOfKafkaNodes; i++) {37            kafkaContainers[i] = new KafkaContainer();38        }39    }40    public void start() {41        for (KafkaContainer kafkaContainer : kafkaContainers) {42            kafkaContainer.start();43        }44    }45    public void stop() {46        for (KafkaContainer kafkaContainer : kafkaContainers) {47            kafkaContainer.stop();48        }49    }50    public String getBootstrapServers() {51        StringBuilder bootstrapServers = new StringBuilder();52        for (KafkaContainer kafkaContainer : kafkaContainers) {53            bootstrapServers.append(kafkaContainer.getBootstrapServers());54            bootstrapServers.append(",");55        }56        return bootstrapServers.toString();57    }58}

start

Using AI Code Generation

1package com.example.kafkacluster;2import org.testcontainers.containers.GenericContainer;3import org.testcontainers.containers.Network;4import org.testcontainers.containers.wait.strategy.Wait;5import java.util.ArrayList;6import java.util.List;7public class KafkaContainerCluster {8    private List<KafkaContainer> kafkaContainers = new ArrayList<>();9    private Network network;10    private int numberOfBrokers;11    public KafkaContainerCluster(int numberOfBrokers) {12        this.numberOfBrokers = numberOfBrokers;13    }14    public void start() {15        network = Network.newNetwork();16        for (int i = 0; i < numberOfBrokers; i++) {17            KafkaContainer kafkaContainer = new KafkaContainer()18                    .withNetwork(network)19                    .withNetworkAliases("kafka-" + i)20                    .withEnv("KAFKA_BROKER_ID", String.valueOf(i))21                    .withEnv("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181")22                    .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")23                    .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")24                    .withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0")25                    .waitingFor(Wait.forLogMessage(".*started.*", 1));26            kafkaContainers.add(kafkaContainer);27            kafkaContainer.start();28        }29    }30    public void stop() {31        kafkaContainers.forEach(GenericContainer::stop);32    }33    public List<KafkaContainer> getKafkaContainers() {34        return kafkaContainers;35    }36    public Network getNetwork() {37        return network;38    }39}40package com.example.kafkacluster;41import org.apache.kafka.clients.admin.AdminClient;42import org.apache.kafka.clients.admin.AdminClientConfig;43import org.apache.kafka.clients.admin.ListTopicsResult;44import org.apache.kafka.clients.admin.NewTopic;45import org.apache.kafka.clients.consumer.ConsumerConfig;46import org.apache.kafka.clients.producer.ProducerConfig;47import org.apache.kafka.common.serialization.StringDeserializer;48import org.apache.kafka.common.serialization.StringSerializer;49import org.junit.After;50import org.junit.Before;51import org.junit.Test;52import org.testcontainers.containers.Network;53import java.util.Arrays;54import java.util.Collections;55import java.util

start

Using AI Code Generation

1package com.example.kafkacluster;2import org.testcontainers.containers.KafkaContainer;3import org.testcontainers.utility.DockerImageName;4import java.util.ArrayList;5import java.util.List;6import java.util.Properties;7public class KafkaContainerCluster {8    public static void main(String[] args) throws Exception {9        List<KafkaContainer> kafkaContainers = new ArrayList<>();10        for (int i = 0; i < 3; i++) {11            KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.2"));12            kafkaContainer.start();13            kafkaContainers.add(kafkaContainer);14        }15        Properties properties = new Properties();16        properties.put("bootstrap.servers", kafkaContainers.get(0).getBootstrapServers());17        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");18        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");19        System.out.println("kafka cluster started");20        Thread.sleep(30000);21        System.out.println("stopping kafka cluster");22        kafkaContainers.forEach(KafkaContainer::stop);23        System.out.println("kafka cluster stopped");24    }25}

start

Using AI Code Generation

1public class Main {2    public static void main(String[] args) throws Exception {3        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(1);4        kafkaContainerCluster.start();5    }6}7public class Main {8    public static void main(String[] args) throws Exception {9        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(3);10        kafkaContainerCluster.start();11    }12}13public class Main {14    public static void main(String[] args) throws Exception {15        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(5);16        kafkaContainerCluster.start();17    }18}19public class Main {20    public static void main(String[] args) throws Exception {21        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(7);22        kafkaContainerCluster.start();23    }24}25public class Main {26    public static void main(String[] args) throws Exception {27        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(9);28        kafkaContainerCluster.start();29    }30}31public class Main {32    public static void main(String[] args) throws Exception {33        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(11);34        kafkaContainerCluster.start();35    }36}37public class Main {38    public static void main(String[] args) throws Exception {39        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(13);40        kafkaContainerCluster.start();41    }42}43public class Main {44    public static void main(String[] args) throws Exception {45        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster(15);46        kafkaContainerCluster.start();47    }48}

start

Using AI Code Generation

1public class KafkaContainerClusterTest {2    public static void main(String[] args) throws Exception {3        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();4        kafkaContainerCluster.start();5    }6}7KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();8kafkaContainerCluster.start();9public class KafkaContainerClusterTest {10    public static void main(String[] args) throws Exception {11        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();12        kafkaContainerCluster.start();13    }14}15KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();16kafkaContainerCluster.start();17public class KafkaContainerClusterTest {18    public static void main(String[] args) throws Exception {19        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();20        kafkaContainerCluster.start();21    }22}23KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();24kafkaContainerCluster.start();25public class KafkaContainerClusterTest {26    public static void main(String[] args) throws Exception {27        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();28        kafkaContainerCluster.start();29    }30}31KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();32kafkaContainerCluster.start();33public class KafkaContainerClusterTest {34    public static void main(String[] args) throws Exception {35        KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();36        kafkaContainerCluster.start();37    }38}39KafkaContainerCluster kafkaContainerCluster = new KafkaContainerCluster();40kafkaContainerCluster.start();

start

Using AI Code Generation

1package com.example.kafkacluster;2import java.util.HashMap;3import java.util.Map;4public class KafkaContainerCluster {5    private static final int ZOOKEEPER_PORT = 2181;6    private static final int KAFKA_PORT = 9092;7    private static final int BROKER_ID = 0;8    private static final String ZOOKEEPER_HOST = "localhost";9    private static final String KAFKA_HOST = "localhost";10    private static final String ZOOKEEPER_CONNECT = "localhost:2181";11    private static final String KAFKA_LISTENER_SECURITY_PROTOCOL_MAP = "PLAINTEXT:PLAINTEXT";12    private static final String KAFKA_INTER_BROKER_LISTENER_NAME = "PLAINTEXT";13    private static final String KAFKA_ZOOKEEPER_CONNECT = "localhost:2181";14    private static final String KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR = "1";15    private static final Map<String, String> ENVIRONMENT_VARIABLES = new HashMap<>();16    static {17        ENVIRONMENT_VARIABLES.put("ZOOKEEPER_PORT", String.valueOf(ZOOKEEPER_PORT));18        ENVIRONMENT_VARIABLES.put("KAFKA_PORT", String.valueOf(KAFKA_PORT));19        ENVIRONMENT_VARIABLES.put("BROKER_ID", String.valueOf(BROKER_ID));20        ENVIRONMENT_VARIABLES.put("ZOOKEEPER_HOST", ZOOKEEPER_HOST);21        ENVIRONMENT_VARIABLES.put("KAFKA_HOST", KAFKA_HOST);22        ENVIRONMENT_VARIABLES.put("ZOOKEEPER_CONNECT", ZOOKEEPER_CONNECT);23        ENVIRONMENT_VARIABLES.put("KAFKA_ADVERTISED_LISTENERS", KAFKA_ADVERTISED_LISTENERS);24        ENVIRONMENT_VARIABLES.put("KAFKA_LISTENERS", KAFKA_LISTENERS);25        ENVIRONMENT_VARIABLES.put("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", KAFKA_LISTENER_SECURITY_PROTOCOL_MAP);26        ENVIRONMENT_VARIABLES.put("KAFKA_INTER_BROKER_LISTENER_NAME", KAFKA_INTER_BROKER_LISTENER_NAME);27        ENVIRONMENT_VARIABLES.put("KAFKA_ZOOKEEPER_CONNECT", KAFKA_ZOOKEEPER_CONNECT);

start

Using AI Code Generation

1package com.example.kafkacluster;2import java.util.Arrays;3import java.util.Properties;4import java.util.Set;5import java.util.concurrent.ExecutionException;6import java.util.concurrent.Future;7import java.util.concurrent.TimeUnit;8import java.util.concurrent.TimeoutException;9import org.apache.kafka.clients.admin.AdminClient;10import org.apache.kafka.clients.admin.AdminClientConfig;11import org.apache.kafka.clients.admin.NewTopic;12import org.apache.kafka.clients.producer.KafkaProducer;13import org.apache.kafka.clients.producer.Producer;14import org.apache.kafka.clients.producer.ProducerRecord;15import org.apache.kafka.clients.producer.RecordMetadata;16import org.apache.kafka.common.serialization.StringSerializer;17import org.slf4j.Logger;18import org.slf4j.LoggerFactory;19public class KafkaContainerCluster {20    private static final Logger log = LoggerFactory.getLogger(KafkaContainerCluster.class);21    private final KafkaContainerWrapper kafkaContainerWrapper;22    public KafkaContainerCluster() {23        this.kafkaContainerWrapper = new KafkaContainerWrapper();24    }25    public void start() {26        kafkaContainerWrapper.start();27    }28    public void stop() {29        kafkaContainerWrapper.stop();30    }31    public void createTopic(String topicName) {32        Properties properties = new Properties();33        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainerWrapper.getBootstrapServers());34        try (AdminClient adminClient = AdminClient.create(properties)) {35            Set<String> existingTopics = adminClient.listTopics().names().get(10, TimeUnit.SECONDS);36            if (!existingTopics.contains(topicName)) {37                NewTopic topic = new NewTopic(topicName, 1, (short) 1);38                adminClient.createTopics(Arrays.asList(topic));39            }40        } catch (InterruptedException | ExecutionException | TimeoutException e) {41            throw new RuntimeException(e);42        }43    }44    public void produce(String topicName, String key, String value) {45        Properties properties = new Properties();46        properties.put("bootstrap.servers", kafkaContainerWrapper.getBootstrapServers());47        properties.put("key.serializer", StringSerializer.class.getName());48        properties.put("value.serializer", StringSerializer.class.getName());49        try (Producer<String, String> producer = new KafkaProducer<>(properties)) {50            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, key, value);51            Future<RecordMetadata> future = producer.send(record);52            future.get(10, TimeUnit.SECONDS);53        } catch (InterruptedException | ExecutionException | TimeoutException

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.