You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/11/23 21:05:01 UTC
[GitHub] [kafka] wcarlson5 commented on a change in pull request #9629: KAFKA-10754: fix flaky tests by waiting kafka streams be in running state before assert
wcarlson5 commented on a change in pull request #9629:
URL: https://github.com/apache/kafka/pull/9629#discussion_r528992986
##########
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:
as above
##########
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:
Yes, Both the old handler test and the close client should have 2 threads. We need to ensure that after a rebalance the old handler has attempted the process the record twice and the client shutdown only once. We can not be sure of that with only one thread.
##########
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");
+ waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION);
Review comment:
The order is not really that important here, either way works
----------------------------------------------------------------
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