counter: Fixed a `ConcurentModificationException`-bug

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 (file)
--- 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<String, String> record) throws JsonProcessingException
+               public synchronized void receive(ConsumerRecord<String, String> 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<Message> getReceivedMessages()
+               {
+                       return received;
+               }
        }
 
        @TestConfiguration