Flaky GlobalStreamThread test
A minor refactor to explicitly verify that Processor#close is only called once.
Reviewers: Guozhang Wang Sophie Blee-Goldman  Bruno Cadonna
@@ -49,14 +49,26 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.assertEquals;
+/**
+ * This test asserts that when Kafka Streams is closing and shuts
+ * down a StreamThread the closing of the GlobalStreamThread happens
+ * after all the StreamThreads are completely stopped.
+ *
+ * The test validates the Processor still has access to the GlobalStateStore while closing.
+ * Otherwise if the GlobalStreamThread were to close underneath the StreamThread
+ * an exception would be thrown as the GlobalStreamThread closes all global stores on closing.
+ */
@Category({IntegrationTest.class})
public class GlobalThreadShutDownOrderTest {
private static final int NUM_BROKERS = 1;
private static final Properties BROKER_CONFIG;
+ private final AtomicInteger closeCounter = new AtomicInteger(0);
+ private final int expectedCloseCount = 1;
static {
BROKER_CONFIG = new Properties();
@@ -136,6 +148,7 @@ public class GlobalThreadShutDownOrderTest {
final List<Long> expectedRetrievedValues = Arrays.asList(1L, 2L, 3L, 4L);
assertEquals(expectedRetrievedValues, retrievedValuesList);
+ assertEquals(expectedCloseCount, closeCounter.get());
}
@@ -188,6 +201,7 @@ public class GlobalThreadShutDownOrderTest {
@Override
public void close() {
+ closeCounter.getAndIncrement();
final List<String> keys = Arrays.asList("A", "B", "C", "D");
for (final String key : keys) {
// need to simulate thread slow in closing
