From 0e1c0c110c1f4f9eb42bea62f9129ae3a7083c81 Mon Sep 17 00:00:00 2001 From: Luke Chen Date: Fri, 20 Nov 2020 15:36:17 +0800 Subject: [PATCH 1/3] KAFKA-10754: fix flaky tests by waiting kafka streams be in running state before assert --- ...caughtExceptionHandlerIntegrationTest.java | 77 ++++++++----------- .../utils/IntegrationTestUtils.java | 13 ++-- 2 files changed, 38 insertions(+), 52 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java index 7747162dbda7d..7cdaa16a374e0 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java @@ -48,7 +48,6 @@ import java.util.Collections; import java.util.List; import java.util.Properties; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.kafka.common.utils.Utils.mkEntry; @@ -68,6 +67,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest { @ClassRule public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1); + public static final Duration DEFAULT_DURATION = Duration.ofSeconds(30); @Rule public TestName testName = new TestName(); @@ -110,26 +110,24 @@ public void teardown() throws IOException { } @Test - public void shouldShutdownThreadUsingOldHandler() throws Exception { + public void shouldShutdownThreadUsingOldHandler() throws InterruptedException { try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) { - final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean flag = new AtomicBoolean(false); kafkaStreams.setUncaughtExceptionHandler((t, e) -> flag.set(true)); StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams); - produceMessages(0L, inputTopic, "A"); - waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, Duration.ofSeconds(15)); TestUtils.waitForCondition(flag::get, "Handler was called"); + waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION); + assertThat(processorValueCollector.size(), equalTo(2)); } } @Test - public void shouldShutdownClient() throws Exception { + public void shouldShutdownClient() throws InterruptedException { try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) { - final CountDownLatch latch = new CountDownLatch(1); kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler")); kafkaStreams.setUncaughtExceptionHandler(exception -> SHUTDOWN_CLIENT); @@ -137,7 +135,7 @@ public void shouldShutdownClient() throws Exception { StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams); produceMessages(0L, inputTopic, "A"); - waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, Duration.ofSeconds(15)); + waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, DEFAULT_DURATION); assertThat(processorValueCollector.size(), equalTo(1)); } @@ -145,48 +143,12 @@ public void shouldShutdownClient() throws Exception { @Test public void shouldShutdownApplication() throws Exception { - final Topology topology = builder.build(); - - try (final KafkaStreams kafkaStreams = new KafkaStreams(topology, properties)) { - final KafkaStreams kafkaStreams1 = new KafkaStreams(topology, properties); - final CountDownLatch latch = new CountDownLatch(1); - kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler")); - kafkaStreams1.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler")); - kafkaStreams.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION); - kafkaStreams1.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION); - - kafkaStreams.start(); - kafkaStreams1.start(); - - produceMessages(0L, inputTopic, "A"); - waitForApplicationState(Arrays.asList(kafkaStreams, kafkaStreams1), KafkaStreams.State.ERROR, Duration.ofSeconds(30)); - - assertThat(processorValueCollector.size(), equalTo(1)); - } + testShutdownApplication(2); } @Test public void shouldShutdownSingleThreadApplication() throws Exception { - properties.setProperty(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "1"); - - final Topology topology = builder.build(); - - try (final KafkaStreams kafkaStreams = new KafkaStreams(topology, properties)) { - final KafkaStreams kafkaStreams1 = new KafkaStreams(topology, properties); - final CountDownLatch latch = new CountDownLatch(1); - kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler")); - kafkaStreams1.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler")); - kafkaStreams.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION); - kafkaStreams1.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION); - - kafkaStreams.start(); - kafkaStreams1.start(); - - produceMessages(0L, inputTopic, "A"); - waitForApplicationState(Arrays.asList(kafkaStreams, kafkaStreams1), KafkaStreams.State.ERROR, Duration.ofSeconds(30)); - - assertThat(processorValueCollector.size(), equalTo(1)); - } + testShutdownApplication(1); } private void produceMessages(final long timestamp, final String streamOneInput, final String msg) { @@ -202,7 +164,6 @@ private void produceMessages(final long timestamp, final String streamOneInput, } private static class ShutdownProcessor extends AbstractProcessor { - final List valueList; ShutdownProcessor(final List valueList) { @@ -215,6 +176,28 @@ public void process(final String key, final String value) { throw new StreamsException(Thread.currentThread().getName()); } } + + private void testShutdownApplication(final int numThreads) throws InterruptedException { + properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numThreads); + + final Topology topology = builder.build(); + + try (final KafkaStreams kafkaStreams1 = new KafkaStreams(topology, properties); + final KafkaStreams kafkaStreams2 = new KafkaStreams(topology, properties)) { + kafkaStreams1.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler")); + kafkaStreams2.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler")); + kafkaStreams1.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION); + kafkaStreams2.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION); + + StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams1); + StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams2); + + produceMessages(0L, inputTopic, "A"); + waitForApplicationState(Arrays.asList(kafkaStreams1, kafkaStreams2), KafkaStreams.State.ERROR, DEFAULT_DURATION); + + assertThat(processorValueCollector.size(), equalTo(1)); + } + } } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java index 03b287d25c236..8ee83cb837cdc 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java @@ -833,7 +833,7 @@ private static void waitUntilMetadataIsPropagated(final List server * {@link State#RUNNING} state at the same time. Note that states may change between the time * that this method returns and the calling function executes its next statement.

* - * When the application is already started use {@link #waitForApplicationState(List, State, Duration)} + * If the application is already started, use {@link #waitForApplicationState(List, State, Duration)} * to wait for instances to reach {@link State#RUNNING} state. * * @param streamsList the list of streams instances to run. @@ -903,16 +903,19 @@ public static void startApplicationAndWaitUntilRunning(final List } /** - * Waits for the given {@link KafkaStreams} instances to all be in a {@link State#RUNNING} - * state. Prefer {@link #startApplicationAndWaitUntilRunning(List, Duration)} when possible + * Waits for the given {@link KafkaStreams} instances to all be in a specific {@link State}. + * Prefer {@link #startApplicationAndWaitUntilRunning(List, Duration)} when possible * because this method uses polling, which can be more error prone and slightly slower. * * @param streamsList the list of streams instances to run. - * @param timeout the time to wait for the streams to all be in {@link State#RUNNING} state. + * @param state the expected state that all the streams to be in within timeout + * @param timeout the time to wait for the streams to all be in the specific state. + * + * @throws InterruptedException if the streams doesn't change to the expected state in time. */ public static void waitForApplicationState(final List streamsList, final State state, - final Duration timeout) throws Exception { + final Duration timeout) throws InterruptedException { retryOnExceptionWithTimeout(timeout.toMillis(), () -> { final Map streamsToStates = streamsList .stream() From e6d39f6dc15a198a5a58d34d239a1021eeaf43b7 Mon Sep 17 00:00:00 2001 From: Luke Chen Date: Sat, 21 Nov 2020 14:58:00 +0800 Subject: [PATCH 2/3] KAFKA-10754: fix shouldShutdownThreadUsingOldHandler by using only 1 stream thread --- .../main/java/org/apache/kafka/streams/KafkaStreams.java | 9 ++++----- .../StreamsUncaughtExceptionHandlerIntegrationTest.java | 8 ++++---- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java index 68f1f3851aa64..4991c93fef788 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -350,7 +350,7 @@ public void setStateListener(final KafkaStreams.StateListener listener) { } /** - * Set the handler invoked when a {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG internal thread} abruptly + * Set the handler invoked when a {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG} internal thread abruptly * terminates due to an uncaught exception. * * @param uncaughtExceptionHandler the uncaught exception handler for all internal threads; {@code null} deletes the current handler @@ -379,13 +379,12 @@ public void setUncaughtExceptionHandler(final Thread.UncaughtExceptionHandler un } /** - * Set the handler invoked when an {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG internal thread} + * Set the handler invoked when an {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG} internal thread * throws an unexpected exception. * These might be exceptions indicating rare bugs in Kafka Streams, or they - * might be exceptions thrown by your code, for example a NullPointerException thrown from your processor - * logic. + * might be exceptions thrown by your code, for example a NullPointerException thrown from your processor logic. * The handler will execute on the thread that produced the exception. - * In order to get the thread that threw the exception, Thread.currentThread(). + * In order to get the thread that threw the exception, use {@code Thread.currentThread()}. *

* Note, this handler must be threadsafe, since it will be shared among all threads, and invoked from any * thread that encounters such an exception. diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java index 7cdaa16a374e0..d24ec1e7ddca7 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java @@ -97,7 +97,7 @@ public void setup() { mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()), mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId), mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()), - mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2), + mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1), mkEntry(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class), mkEntry(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class) ) @@ -121,7 +121,7 @@ public void shouldShutdownThreadUsingOldHandler() throws InterruptedException { TestUtils.waitForCondition(flag::get, "Handler was called"); waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION); - assertThat(processorValueCollector.size(), equalTo(2)); + assertThat(processorValueCollector.size(), equalTo(1)); } } @@ -142,12 +142,12 @@ public void shouldShutdownClient() throws InterruptedException { } @Test - public void shouldShutdownApplication() throws Exception { + public void shouldShutdownMultipleThreadApplication() throws InterruptedException { testShutdownApplication(2); } @Test - public void shouldShutdownSingleThreadApplication() throws Exception { + public void shouldShutdownSingleThreadApplication() throws InterruptedException { testShutdownApplication(1); } From 2b6d0a2d285b5b7fc0a9a8474712870f6f7a767e Mon Sep 17 00:00:00 2001 From: Luke Chen Date: Tue, 24 Nov 2020 12:26:39 +0800 Subject: [PATCH 3/3] KAFKA-10754: revert to set the default thread number as 2, and make the test more reliable --- .../org/apache/kafka/streams/KafkaStreams.java | 4 ++-- ...sUncaughtExceptionHandlerIntegrationTest.java | 16 ++++++++++------ 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java index 4991c93fef788..e0f079612ce44 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -350,7 +350,7 @@ public void setStateListener(final KafkaStreams.StateListener listener) { } /** - * Set the handler invoked when a {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG} internal thread abruptly + * Set the handler invoked when an internal {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG stream thread} abruptly * terminates due to an uncaught exception. * * @param uncaughtExceptionHandler the uncaught exception handler for all internal threads; {@code null} deletes the current handler @@ -379,7 +379,7 @@ public void setUncaughtExceptionHandler(final Thread.UncaughtExceptionHandler un } /** - * Set the handler invoked when an {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG} internal thread + * Set the handler invoked when an internal {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG stream thread} * throws an unexpected exception. * These might be exceptions indicating rare bugs in Kafka Streams, or they * might be exceptions thrown by your code, for example a NullPointerException thrown from your processor logic. diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java index d24ec1e7ddca7..c7d9308e48d15 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java @@ -48,7 +48,7 @@ import java.util.Collections; import java.util.List; import java.util.Properties; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; @@ -97,7 +97,7 @@ public void setup() { mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()), mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId), mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()), - mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1), + mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2), mkEntry(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class), mkEntry(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class) ) @@ -112,16 +112,20 @@ public void teardown() throws IOException { @Test public void shouldShutdownThreadUsingOldHandler() throws InterruptedException { try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) { - final AtomicBoolean flag = new AtomicBoolean(false); - kafkaStreams.setUncaughtExceptionHandler((t, e) -> flag.set(true)); + final AtomicInteger counter = new AtomicInteger(0); + kafkaStreams.setUncaughtExceptionHandler((t, e) -> counter.incrementAndGet()); StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams); produceMessages(0L, inputTopic, "A"); - TestUtils.waitForCondition(flag::get, "Handler was called"); + // should call the UncaughtExceptionHandler in current thread + TestUtils.waitForCondition(() -> counter.get() == 1, "Handler was called 1st time"); + // should call the UncaughtExceptionHandler after rebalancing to another thread + TestUtils.waitForCondition(() -> counter.get() == 2, DEFAULT_DURATION.toMillis(), "Handler was called 2nd time"); + // the stream should now turn into ERROR state after 2 threads are dead waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION); - assertThat(processorValueCollector.size(), equalTo(1)); + assertThat(processorValueCollector.size(), equalTo(2)); } }