From 5826111b853d2ae4a2e6f061ae1a079ad524da89 Mon Sep 17 00:00:00 2001 From: Mahmoud Ben Hassine Date: Tue, 17 Nov 2020 16:18:05 +0100 Subject: [PATCH] Fix KafkaItemReaderTests Tests in this class fail intermittently because they send messages to Kafka in an asynchronous way and assert on the results immediately without waiting for the send operation to complete. This commit updates the tests to wait for send results before asserting on them (similar to a140a9f5). --- .../item/kafka/KafkaItemReaderTests.java | 69 ++++++++++++------- 1 file changed, 44 insertions(+), 25 deletions(-) diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java index 72458b9f1..b232b617a 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java @@ -38,6 +38,7 @@ import org.springframework.batch.item.ExecutionContext; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.SendResult; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.util.concurrent.ListenableFuture; @@ -186,12 +187,16 @@ public class KafkaItemReaderTests { } @Test - public void testReadFromSinglePartition() { + public void testReadFromSinglePartition() throws ExecutionException, InterruptedException { this.template.setDefaultTopic("topic1"); - this.template.sendDefault("val0"); - this.template.sendDefault("val1"); - this.template.sendDefault("val2"); - this.template.sendDefault("val3"); + List>> futures = new ArrayList<>(); + futures.add(this.template.sendDefault("val0")); + futures.add(this.template.sendDefault("val1")); + futures.add(this.template.sendDefault("val2")); + futures.add(this.template.sendDefault("val3")); + for (ListenableFuture> future : futures) { + future.get(); + } this.reader = new KafkaItemReader<>(this.consumerProperties, "topic1", 0); this.reader.setPollTimeout(Duration.ofSeconds(1)); @@ -216,12 +221,16 @@ public class KafkaItemReaderTests { } @Test - public void testReadFromSinglePartitionFromCustomOffset() { + public void testReadFromSinglePartitionFromCustomOffset() throws ExecutionException, InterruptedException { this.template.setDefaultTopic("topic5"); - this.template.sendDefault("val0"); // <-- offset 0 - this.template.sendDefault("val1"); // <-- offset 1 - this.template.sendDefault("val2"); // <-- offset 2 - this.template.sendDefault("val3"); // <-- offset 3 + List>> futures = new ArrayList<>(); + futures.add(this.template.sendDefault("val0")); // <-- offset 0 + futures.add(this.template.sendDefault("val1")); // <-- offset 1 + futures.add(this.template.sendDefault("val2")); // <-- offset 2 + futures.add(this.template.sendDefault("val3")); // <-- offset 3 + for (ListenableFuture> future : futures) { + future.get(); + } this.reader = new KafkaItemReader<>(this.consumerProperties, "topic5", 0); @@ -250,9 +259,12 @@ public class KafkaItemReaderTests { // first run: read a topic from the beginning this.template.setDefaultTopic("topic6"); - this.template.sendDefault("val0"); // <-- offset 0 - this.template.sendDefault("val1"); // <-- offset 1 - + List>> futures = new ArrayList<>(); + futures.add(this.template.sendDefault("val0")); // <-- offset 0 + futures.add(this.template.sendDefault("val1")); // <-- offset 1 + for (ListenableFuture> future : futures) { + future.get(); + } this.reader = new KafkaItemReader<>(this.consumerProperties, "topic6", 0); this.reader.setPollTimeout(Duration.ofSeconds(1)); this.reader.open(new ExecutionContext()); @@ -299,12 +311,16 @@ public class KafkaItemReaderTests { } @Test - public void testReadFromMultiplePartitions() { + public void testReadFromMultiplePartitions() throws ExecutionException, InterruptedException { this.template.setDefaultTopic("topic2"); - this.template.sendDefault("val0"); - this.template.sendDefault("val1"); - this.template.sendDefault("val2"); - this.template.sendDefault("val3"); + List>> futures = new ArrayList<>(); + futures.add(this.template.sendDefault("val0")); + futures.add(this.template.sendDefault("val1")); + futures.add(this.template.sendDefault("val2")); + futures.add(this.template.sendDefault("val3")); + for (ListenableFuture> future : futures) { + future.get(); + } this.reader = new KafkaItemReader<>(this.consumerProperties, "topic2", 0, 1); this.reader.setPollTimeout(Duration.ofSeconds(1)); @@ -323,14 +339,17 @@ public class KafkaItemReaderTests { } @Test - public void testReadFromSinglePartitionAfterRestart() { + public void testReadFromSinglePartitionAfterRestart() throws ExecutionException, InterruptedException { this.template.setDefaultTopic("topic3"); - this.template.sendDefault("val0"); - this.template.sendDefault("val1"); - this.template.sendDefault("val2"); - this.template.sendDefault("val3"); - this.template.sendDefault("val4"); - + List>> futures = new ArrayList<>(); + futures.add(this.template.sendDefault("val0")); + futures.add(this.template.sendDefault("val1")); + futures.add(this.template.sendDefault("val2")); + futures.add(this.template.sendDefault("val3")); + futures.add(this.template.sendDefault("val4")); + for (ListenableFuture> future : futures) { + future.get(); + } ExecutionContext executionContext = new ExecutionContext(); Map offsets = new HashMap<>(); offsets.put(new TopicPartition("topic3", 0), 1L);