From a140a9f5a14b250af94ea4624501068b647b4bbb Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Fri, 2 Aug 2019 17:06:54 -0500 Subject: [PATCH] Attempting to fix the KafkaItemReaderTests#testReadFromMultiplePartitionsAfterRestart --- .../item/kafka/KafkaItemReaderTests.java | 25 ++++++++++++------- 1 file changed, 16 insertions(+), 9 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 6e419e689..75e256353 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 @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.ExecutionException; import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -38,6 +39,7 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.util.concurrent.ListenableFuture; import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.is; @@ -269,15 +271,20 @@ public class KafkaItemReaderTests { } @Test - public void testReadFromMultiplePartitionsAfterRestart() { - this.template.send("topic4", 0, null, "val0"); - this.template.send("topic4", 0, null, "val2"); - this.template.send("topic4", 0, null, "val4"); - this.template.send("topic4", 0, null, "val6"); - this.template.send("topic4", 1, null, "val1"); - this.template.send("topic4", 1, null, "val3"); - this.template.send("topic4", 1, null, "val5"); - this.template.send("topic4", 1, null, "val7"); + public void testReadFromMultiplePartitionsAfterRestart() throws ExecutionException, InterruptedException { + List futures = new ArrayList<>(); + futures.add(this.template.send("topic4", 0, null, "val0")); + futures.add(this.template.send("topic4", 0, null, "val2")); + futures.add(this.template.send("topic4", 0, null, "val4")); + futures.add(this.template.send("topic4", 0, null, "val6")); + futures.add(this.template.send("topic4", 1, null, "val1")); + futures.add(this.template.send("topic4", 1, null, "val3")); + futures.add(this.template.send("topic4", 1, null, "val5")); + futures.add(this.template.send("topic4", 1, null, "val7")); + + for (ListenableFuture future : futures) { + future.get(); + } ExecutionContext executionContext = new ExecutionContext(); Map offsets = new HashMap<>();