From fa2e0230ab014b8030acde4c477a5e5a919c0c0b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 12 Dec 2017 17:41:57 -0500 Subject: [PATCH] Declare topics in KafkaMessageListenerContTests Since JSON tests expect an assignments as 2 partitions per topic, it would be better to do that explicitly via `KafkaEmbedded` rather than rely on the auto-topic creation which might not bring us 2 anticipated partitions --- .../KafkaMessageListenerContainerTests.java | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 7b91cc39..f0ff0651 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -89,7 +89,6 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; * @author Martin Dam * @author Artem Bilan * @author Loic Talhouarne - * @author Tom van den Berge */ public class KafkaMessageListenerContainerTests { @@ -131,12 +130,11 @@ public class KafkaMessageListenerContainerTests { private static String topic18 = "testTopic18"; - private static String topic19 = "testTopic19"; @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic5, - topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17, topic18, - topic19); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5, + topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17, + topic18); @Rule public TestName testName = new TestName(); @@ -1618,6 +1616,7 @@ public class KafkaMessageListenerContainerTests { consumer.close(); logger.info("Stop rebalance after failed record"); } + private Consumer spyOnConsumer(KafkaMessageListenerContainer container) { Consumer consumer = spy( KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class)); @@ -1626,10 +1625,8 @@ public class KafkaMessageListenerContainerTests { return consumer; } - private KafkaMessageListenerContainer spyOnContainer( - KafkaMessageListenerContainer container, + private KafkaMessageListenerContainer spyOnContainer(KafkaMessageListenerContainer container, final CountDownLatch stubbingComplete) { - KafkaMessageListenerContainer spy = spy(container); willAnswer(i -> { if (stubbingComplete.getCount() > 0 && Thread.currentThread().getName().endsWith("-C-1")) {