How to use testKafkaFunctionality method of org.testcontainers.containers.KafkaContainerTest class

Best Testcontainers-java code snippet using org.testcontainers.containers.KafkaContainerTest.testKafkaFunctionality

Source:KafkaContainerTest.java

...39    @Test40    public void testUsage() throws Exception {41        try (KafkaContainer kafka = new KafkaContainer(KAFKA_TEST_IMAGE)) {42            kafka.start();43            testKafkaFunctionality(kafka.getBootstrapServers());44        }45    }46    @Test47    public void testWithHostExposedPort() throws Exception {48        Testcontainers.exposeHostPorts(12345);49        try (KafkaContainer kafka = new KafkaContainer(KAFKA_TEST_IMAGE)) {50            kafka.start();51            testKafkaFunctionality(kafka.getBootstrapServers());52        }53    }54    @Test55    public void testExternalZookeeperWithExternalNetwork() throws Exception {56        try (57                Network network = Network.newNetwork();58                // withExternalZookeeper {59                KafkaContainer kafka = new KafkaContainer(KAFKA_TEST_IMAGE)60                        .withNetwork(network)61                        .withExternalZookeeper("zookeeper:2181");62                // }63                GenericContainer<?> zookeeper = new GenericContainer<>(ZOOKEEPER_TEST_IMAGE)64                        .withNetwork(network)65                        .withNetworkAliases("zookeeper")66                        .withEnv("ZOOKEEPER_CLIENT_PORT", "2181");67                // withKafkaNetwork {68                GenericContainer<?> application = new GenericContainer<>(DockerImageName.parse("alpine"))69                        .withNetwork(network)70                        // }71                        .withNetworkAliases("dummy")72                        .withCommand("sleep 10000")73        ) {74            zookeeper.start();75            kafka.start();76            application.start();77            testKafkaFunctionality(kafka.getBootstrapServers());78        }79    }80    protected void testKafkaFunctionality(String bootstrapServers) throws Exception {81        testKafkaFunctionality(bootstrapServers, 1, 1);82    }83    protected void testKafkaFunctionality(String bootstrapServers, int partitions, int rf) throws Exception {84        try (85                AdminClient adminClient = AdminClient.create(86                        ImmutableMap.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)87                );88                KafkaProducer<String, String> producer = new KafkaProducer<>(89                        ImmutableMap.of(90                                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,91                                bootstrapServers,92                                ProducerConfig.CLIENT_ID_CONFIG,93                                UUID.randomUUID().toString()94                        ),95                        new StringSerializer(),96                        new StringSerializer()97                );...

testKafkaFunctionality

Using AI Code Generation

1import org.testcontainers.containers.KafkaContainer2import org.testcontainers.utility.DockerImageName3import org.testcontainers.containers.Network4import org.testcontainers.containers.GenericContainer5import org.testcontainers.containers.output.Slf4jLogConsumer6import org.slf4j.Logger7import org.slf4j.LoggerFactory8import java.time.Duration9import java.util.concurrent.TimeUnit10import java.util.concurrent.TimeoutException11import java.util.concurrent.atomic.AtomicBoolean12import java.util.stream.Collectors13import org.apache.kafka.clients.consumer.ConsumerConfig14import org.apache.kafka.clients.consumer.ConsumerRecord15import org.apache.kafka.clients.consumer.ConsumerRecords16import org.apache.kafka.clients.consumer.KafkaConsumer17import org.apache.kafka.clients.producer.KafkaProducer18import org.apache.kafka.clients.producer.ProducerConfig19import org.apache.kafka.clients.producer.ProducerRecord20import org.apache.kafka.common.TopicPartition21import org.testcontainers.containers.KafkaContainerTest22import org.testcontainers.containers.KafkaContainerTest.Companion.KAFKA_PORT23import org.testcontainers.containers.KafkaContainerTest.Companion.ZOOKEEPER_PORT24import org.testcontainers.containers.KafkaContainerTest.Companion.BOOTSTRAP_SERVERS25import org.testcontainers.containers.KafkaContainerTest.Companion.ZOOKEEPER_CONNECT26import org.testcontainers.containers.KafkaContainerTest.Companion.TOPIC27import org.testcontainers.containers.KafkaContainerTest.Companion.MESSAGE28import org.testcontainers.containers.KafkaContainerTest.Companion.KEY_SERIALIZER29import org.testcontainers.containers.KafkaContainerTest.Companion.VALUE_SERIALIZER30import org.testcontainers.containers.KafkaContainerTest.Companion.KEY_DESERIALIZER31import org.testcontainers.containers.KafkaContainerTest.Companion.VALUE_DESERIALIZER32import org.testcontainers.containers.KafkaContainerTest.Companion.KEY33import org.testcontainers.containers.KafkaContainerTest.Companion.VALUE34import org.testcontainers.containers.KafkaContainerTest.Companion.CONSUMER_GROUP_ID35import org.testcontainers.containers.KafkaContainerTest.Companion.DEFAULT_TIMEOUT_SECONDS36import org.testcontainers.containers.KafkaContainerTest.Companion.createConsumerProperties37import org.testcontainers.containers.KafkaContainerTest.Companion.createProducerProperties38import org.testcontainers.containers.KafkaContainerTest.Companion.createConsumer39import org.testcontainers.containers.KafkaContainerTest.Companion.createProducer40import org.testcontainers.containers.KafkaContainerTest.Companion.createConsumerRecord41import org.testcontainers.containers.KafkaContainerTest.Companion.createConsumerRecords42import org.testcontainers.containers.K

testKafkaFunctionality

Using AI Code Generation

1    public void testKafkaFunctionality() throws Exception {2        String topic = "test-topic";3        String message = "test-message";4        try (KafkaContainer kafka = new KafkaContainer()) {5            kafka.start();6            try (ZookeeperContainer zookeeper = new ZookeeperContainer()) {7                zookeeper.start();8                try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafka.getConsumerProperties(zookeeper.getConnectString(), "test-group", StringDeserializer.class, StringDeserializer.class))) {9                    consumer.subscribe(Collections.singleton(topic));10                    try (KafkaProducer<String, String> producer = new KafkaProducer<>(kafka.getProducerProperties())) {11                        producer.send(new ProducerRecord<>(topic, message));12                        producer.flush();13                        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));14                        assertEquals(1, records.count());15                        assertEquals(message, records.iterator().next().value());16                    }17                }18            }19        }20    }21    public void testKafkaFunctionality() throws Exception {22        String topic = "test-topic";23        String message = "test-message";24        try (KafkaContainer kafka = new KafkaContainer()) {25            kafka.start();26            try (ZookeeperContainer zookeeper = new ZookeeperContainer()) {27                zookeeper.start();28                try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafka.getConsumerProperties(zookeeper.getConnectString(), "test-group", StringDeserializer.class, StringDeserializer.class))) {29                    consumer.subscribe(Collections.singleton(topic));30                    try (KafkaProducer<String, String> producer = new KafkaProducer<>(kafka.getProducerProperties())) {31                        producer.send(new ProducerRecord<>(topic, message));32                        producer.flush();33                        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));34                        assertEquals(1, records.count());35                        assertEquals(message, records.iterator().next().value());36                    }37                }38            }39        }40    }41    public void testKafkaFunctionality() throws Exception {42        String topic = "test-topic";43        String message = "test-message";44        try (KafkaContainer kafka = new KafkaContainer()) {45            kafka.start();46            try (Zookeeper

testKafkaFunctionality

Using AI Code Generation

1    void testKafkaFunctionality() throws InterruptedException {2        try (KafkaContainer kafka = new KafkaContainer()) {3            kafka.start();4            String topic = "test-topic";5            String message = "hello world";6            String broker = kafka.getBootstrapServers();7            try (AdminClient adminClient = AdminClient.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker))) {8                adminClient.createTopics(List.of(new NewTopic(topic, 1, (short) 1))).all().get();9            }10            try (KafkaProducer<String, String> producer = new KafkaProducer<>(Map.of(11            ))) {12                producer.send(new ProducerRecord<>(topic, message)).get();13            }14            try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(Map.of(15            ))) {16                consumer.subscribe(List.of(topic));17                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));18                assertThat(records.count()).isEqualTo(1);19                assertThat(records.iterator().next().value()).isEqualTo(message);20            }21        }22    }23}24The testKafkaFunctionality() method

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.