[GitHub] [kafka] ableegoldman commented on a change in pull request #9629: KAFKA-10754: fix flaky tests by waiting kafka streams be in running state before assert

[GitBox <gi...@apache.org>]

ableegoldman commented on a change in pull request #9629:
URL: https://github.com/apache/kafka/pull/9629#discussion_r528943389



##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -110,83 +110,45 @@ 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");
-            assertThat(processorValueCollector.size(), equalTo(2));

Review comment:
       @wcarlson5 for example, this test probably should have multiple threads, right?

##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -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

Review comment:
       I think this was actually correct as it was (and ditto for the above). One alternative suggestion:
   
   ```suggestion
        * Set the handler invoked when an internal {@link StreamsConfig#NUM_STREAM_THREADS_CONFIG stream thread}
   ```

##########
File path: 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),

Review comment:
       Hey @wcarlson5 , can you take a look at this? If we change the default number of threads to 1 will we be reducing test coverage or not testing the correct thing anymore?
   
   FWIW I think for tests where the number of threads doesn't matter, we should default to 1. But I'm not sure which tests do/do not rely on using multiple stream threads




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org