From 50c9fa713a34aacfb846077520aba2c043e692d9 Mon Sep 17 00:00:00 2001 From: Bill Bejeck Date: Wed, 27 Feb 2019 12:30:08 -0500 Subject: [PATCH] KAFKA-8011: Fix for race condition causing concurrent modification exception --- .../streams/integration/RegexSourceIntegrationTest.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java index 264cd35e45733..1cb9e0c934d47 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java @@ -60,6 +60,7 @@ import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; import java.util.regex.Pattern; @@ -152,7 +153,7 @@ public void testRegexMatchesTopicsAWhenCreated() throws Exception { final KStream pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d")); pattern1Stream.to(DEFAULT_OUTPUT_TOPIC, Produced.with(stringSerde, stringSerde)); - final List assignedTopics = new ArrayList<>(); + final List assignedTopics = new CopyOnWriteArrayList<>(); streams = new KafkaStreams(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() { @Override public Consumer getConsumer(final Map config) { @@ -190,7 +191,7 @@ public void testRegexMatchesTopicsAWhenDeleted() throws Exception { pattern1Stream.to(DEFAULT_OUTPUT_TOPIC, Produced.with(stringSerde, stringSerde)); - final List assignedTopics = new ArrayList<>(); + final List assignedTopics = new CopyOnWriteArrayList<>(); streams = new KafkaStreams(builder.build(), streamsConfiguration, new DefaultKafkaClientSupplier() { @Override public Consumer getConsumer(final Map config) {