Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use Docker for integration tests related to Kafka support

Browse Source 
Co-authored-by: patbaumgartner <contact@patbaumgartner.com>
Signed-off-by: martinfrancois <f.martin@fastmail.com>
This commit is contained in:
martinfrancois
2025-06-13 11:31:36 +02:00
committed by Mahmoud Ben Hassine
parent d95397faf0
commit 3eaf8d9ac9
2 changed files with 31 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-batch-infrastructure/pom.xml
View File

@@ -425,6 +425,12 @@
<version>${sqlserver.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<version>${testcontainers.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.thoughtworks.xstream</groupId>
<artifactId>xstream</artifactId>

40
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderIntegrationTests.java
View File

@@ -25,6 +25,8 @@ import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
@@ -36,14 +38,15 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.utility.DockerImageName;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -54,13 +57,17 @@ import static org.junit.jupiter.api.Assertions.assertNull;
/**
* @author Mathieu Ouellet
* @author Mahmoud Ben Hassine
* @author François Martin
* @author Patrick Baumgartner
*/
@EmbeddedKafka
@Testcontainers(disabledWithoutDocker = true)
@ExtendWith(SpringExtension.class)
class KafkaItemReaderIntegrationTests {
@Autowired
private EmbeddedKafkaBroker embeddedKafka;
private static final DockerImageName KAFKA_IMAGE = DockerImageName.parse("apache/kafka:3.9.1");
@Container
public static KafkaContainer kafka = new KafkaContainer(KAFKA_IMAGE);
private KafkaItemReader<String, String> reader;
@@ -69,21 +76,24 @@ class KafkaItemReaderIntegrationTests {
private Properties consumerProperties;
@BeforeAll
static void setUpTopics(@Autowired EmbeddedKafkaBroker embeddedKafka) {
embeddedKafka.addTopics(new NewTopic("topic1", 1, (short) 1), new NewTopic("topic2", 2, (short) 1),
new NewTopic("topic3", 1, (short) 1), new NewTopic("topic4", 2, (short) 1),
new NewTopic("topic5", 1, (short) 1), new NewTopic("topic6", 1, (short) 1));
static void setUpTopics() {
Properties properties = new Properties();
properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
try (AdminClient adminClient = AdminClient.create(properties)) {
adminClient.createTopics(List.of(new NewTopic("topic1", 1, (short) 1), new NewTopic("topic2", 2, (short) 1),
new NewTopic("topic3", 1, (short) 1), new NewTopic("topic4", 2, (short) 1),
new NewTopic("topic5", 1, (short) 1), new NewTopic("topic6", 1, (short) 1)));
}
}
@BeforeEach
void setUp() {
Map<String, Object> producerProperties = KafkaTestUtils.producerProps(embeddedKafka);
Map<String, Object> producerProperties = KafkaTestUtils.producerProps(kafka.getBootstrapServers());
ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerProperties);
this.template = new KafkaTemplate<>(producerFactory);
this.consumerProperties = new Properties();
this.consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
embeddedKafka.getBrokersAsString());
this.consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
this.consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "1");
this.consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
@@ -186,8 +196,8 @@ class KafkaItemReaderIntegrationTests {
this.reader.close();
// The offset stored in Kafka should be equal to 2 at this point
OffsetAndMetadata currentOffset = KafkaTestUtils.getCurrentOffset(embeddedKafka.getBrokersAsString(), "1",
"topic6", 0);
OffsetAndMetadata currentOffset = KafkaTestUtils.getCurrentOffset(kafka.getBootstrapServers(), "1", "topic6",
0);
assertEquals(2, currentOffset.offset());
// second run (with same consumer group ID): new messages arrived since the last