From 7768a8989ee4c1a9340f2e2e396aa7335b37fb0a Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Mon, 27 May 2024 21:25:55 +0200 Subject: [PATCH] counter: 1.2.15 - `ToplogyTestDriver.close` must always be called --- .../CounterStreamProcessorTopologyTest.java | 29 +++++++++++++++---- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java index 16cf5d2..1b3e1e4 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java @@ -6,6 +6,8 @@ import org.apache.kafka.streams.TestOutputTopic; import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.state.Stores; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerde; @@ -28,8 +30,14 @@ public class CounterStreamProcessorTopologyTest public final static String IN = "TEST-IN"; public final static String OUT = "TEST-OUT"; - @Test - public void test() + + TopologyTestDriver testDriver; + TestInputTopic in; + TestOutputTopic out; + + + @BeforeEach + public void setUpTestDriver() { Topology topology = CounterStreamProcessor.buildTopology( IN, @@ -47,18 +55,23 @@ public class CounterStreamProcessorTopologyTest JsonSerde valueSerde = new JsonSerde<>(); valueSerde.configure(propertyMap, false); - TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties); + testDriver = new TopologyTestDriver(topology, streamProcessorProperties); - TestInputTopic in = testDriver.createInputTopic( + in = testDriver.createInputTopic( IN, (JsonSerializer)keySerde.serializer(), (JsonSerializer)valueSerde.serializer()); - TestOutputTopic out = testDriver.createOutputTopic( + out = testDriver.createOutputTopic( OUT, (JsonDeserializer)keySerde.deserializer(), (JsonDeserializer)valueSerde.deserializer()); + } + + @Test + public void test() + { TestData.injectInputMessages((key, value) -> in.pipeInput(key, value)); MultiValueMap receivedMessages = new LinkedMultiValueMap<>(); @@ -77,4 +90,10 @@ public class CounterStreamProcessorTopologyTest TestData.assertExpectedMessages(receivedMessages); } + + @AfterEach + public void tearDown() + { + testDriver.close(); + } } -- 2.20.1