close all kafkaStreams before purgeLocalStreamsState
The flaky tests are because we forgot to close the kafkaStreams before purgeLocalStreamsState, so that sometimes there will be some tmp files be created/deleted during streams running(ex: checkpoint.tmp), and caused the DirectoryNotEmptyException or NoSuchFileException be thrown.
Reviewers:  Levani Kokhreidze, Bill Bejeck
@@ -75,6 +75,7 @@ public class StreamTableJoinTopologyOptimizationIntegrationTest {
private String inputTopic;
private String outputTopic;
private String applicationId;
+ private KafkaStreams kafkaStreams;
private Properties streamsConfiguration;
@@ -119,6 +120,9 @@ public class StreamTableJoinTopologyOptimizationIntegrationTest {
@After
public void whenShuttingDown() throws IOException {
+ if (kafkaStreams != null) {
+ kafkaStreams.close();
+ }
IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
}
@@ -137,7 +141,7 @@ public class StreamTableJoinTopologyOptimizationIntegrationTest {
.join(table, (value1, value2) -> value2)
.to(outputTopic);
- startStreams(streamsBuilder);
+ kafkaStreams = startStreams(streamsBuilder);
final long timestamp = System.currentTimeMillis();
@@ -149,8 +153,6 @@ public class StreamTableJoinTopologyOptimizationIntegrationTest {
sendEvents(inputTopic, timestamp, expectedRecords);
sendEvents(outputTopic, timestamp, expectedRecords);
- startStreams(streamsBuilder);
-
validateReceivedMessages(
outputTopic,
new IntegerDeserializer(),
