diff --git a/modules/kafka/src/test/java/org/testcontainers/AbstractKafka.java b/modules/kafka/src/test/java/org/testcontainers/AbstractKafka.java index 90977e61c18..5a5692a336e 100644 --- a/modules/kafka/src/test/java/org/testcontainers/AbstractKafka.java +++ b/modules/kafka/src/test/java/org/testcontainers/AbstractKafka.java @@ -141,6 +141,63 @@ protected void testKafkaFunctionality( } } + protected void testKafkaTransactionalFunctionality(String bootstrapServers) throws Exception { + Properties adminClientProperties = new Properties(); + adminClientProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + + Properties consumerProperties = new Properties(); + consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "tc-" + UUID.randomUUID()); + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProperties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); + + Properties producerProperties = new Properties(); + producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + producerProperties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tc-" + UUID.randomUUID()); + // Fail fast if the transaction state topic can not be created + producerProperties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000"); + + try ( + AdminClient adminClient = AdminClient.create(adminClientProperties); + KafkaProducer producer = new KafkaProducer<>( + producerProperties, + new StringSerializer(), + new StringSerializer() + ); + KafkaConsumer consumer = new KafkaConsumer<>( + consumerProperties, + new StringDeserializer(), + new StringDeserializer() + ); + ) { + String topicName = "messages-" + UUID.randomUUID(); + + Collection topics = Collections.singletonList(new NewTopic(topicName, 1, (short) 1)); + adminClient.createTopics(topics).all().get(30, TimeUnit.SECONDS); + + consumer.subscribe(Collections.singletonList(topicName)); + + producer.initTransactions(); + producer.beginTransaction(); + producer.send(new ProducerRecord<>(topicName, "testcontainers", "rulezzz")).get(); + producer.commitTransaction(); + + Awaitility + .await() + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + + assertThat(records) + .hasSize(1) + .extracting(ConsumerRecord::topic, ConsumerRecord::key, ConsumerRecord::value) + .containsExactly(tuple(topicName, "testcontainers", "rulezzz")); + }); + + consumer.unsubscribe(); + } + } + protected static String getJaasConfig() { String jaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required " + diff --git a/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java b/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java index 1c95bf95e02..c9fdb3db1d4 100644 --- a/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java +++ b/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java @@ -24,6 +24,14 @@ void testUsage() throws Exception { } } + @Test + void testUsageWithTransactions() throws Exception { + try (ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.4.0")) { + kafka.start(); + testKafkaTransactionalFunctionality(kafka.getBootstrapServers()); + } + } + @Test void testUsageWithListener() throws Exception { try ( diff --git a/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java b/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java index 34af2e1784d..af56ff1ab9e 100644 --- a/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java +++ b/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java @@ -21,6 +21,14 @@ void testUsage() throws Exception { } } + @Test + void testUsageWithTransactions() throws Exception { + try (KafkaContainer kafka = new KafkaContainer("apache/kafka-native:3.8.0")) { + kafka.start(); + testKafkaTransactionalFunctionality(kafka.getBootstrapServers()); + } + } + @Test void testUsageWithListener() throws Exception { try (