From 145d5414195ca3f536deed73975a6af80bfe99ba Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 11 Feb 2023 14:27:15 +0100 Subject: [PATCH] counter: Fixed a `ConcurentModificationException`-bug --- .../kafka/wordcount/counter/CounterApplicationIT.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java index b314726..a345935 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java @@ -65,7 +65,7 @@ public class CounterApplicationIT await("Expexted converted data") .atMost(Duration.ofSeconds(10)) - .untilAsserted(() -> TestData.assertExpectedResult(consumer.received)); + .untilAsserted(() -> TestData.assertExpectedResult(consumer.getReceivedMessages())); } @@ -76,12 +76,17 @@ public class CounterApplicationIT private final ObjectMapper mapper; @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) - public void receive(ConsumerRecord record) throws JsonProcessingException + public synchronized void receive(ConsumerRecord record) throws JsonProcessingException { log.debug("Received message: {}", record); Key key = mapper.readValue(record.key(), Key.class); received.add(Message.of(key,record.value())); } + + synchronized List getReceivedMessages() + { + return received; + } } @TestConfiguration -- 2.20.1