KAFKA-10754: fix flaky tests by waiting kafka streams be in running state before assert by showuon · Pull Request #9629 · apache/kafka

showuon · 2020-11-20T07:44:52Z

The flaky test is because we didn't wait for the streams become RUNNING before verifying the state becoming ERROR state. This fix explicitly wait for the streams become RUNNING state. Also, I found we didn't put the 2nd stream into try resource block, so it won't close the stream after tests. In addition, I also fix some logic errors in the tests.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

showuon · 2020-11-20T07:45:32Z

forgot to put the kafkaStreams1 into the try resource block here

showuon · 2020-11-20T07:46:44Z

The start is async, and we didn't wait for it.

showuon · 2020-11-20T07:48:25Z

@ableegoldman @wcarlson5 , could you help review this PR? Thanks.

showuon · 2020-11-20T14:16:08Z

delete unused latch

showuon · 2020-11-20T14:17:03Z

We should throw a specific kind of exception, not an Exception.

…tate before assert

showuon · 2020-11-20T14:49:19Z

-            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);


We should wait for the uncaughtExceptionHandler got called before waiting for the streams state change.

The order is not really that important here, either way works

ableegoldman · 2020-11-20T19:15:27Z

Hey @showuon thanks for the quick fix! I notice that StreamsUncaughtExceptionHandlerIntegrationTest.shouldShutdownThreadUsingOldHandler still failed on the JDK15 PR build, can you look into that?

showuon · 2020-11-21T07:04:54Z

@ableegoldman , thanks for pointing it out. After investigation, I found the test StreamsUncaughtExceptionHandlerIntegrationTest.shouldShutdownThreadUsingOldHandler failed is because we set 2 stream threads for this test. So when we got the uncaughtException, we shutdown the thread, and rebalancing to the other thread. And we have to wait for rebalancing completes, and later another exception thrown in the thread, then the stream will turn into ERROR state, which is why it is so flaky.

I default set to 1 stream thread in this test since other tests will set to the expected thread number before testing.

The fix is in this commit: e6d39f6. Thank you.

…stream thread

showuon · 2020-11-21T10:07:29Z

All tests passed, Yeah!

ableegoldman · 2020-11-23T19:21:42Z


    /**
-     * 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


I think this was actually correct as it was (and ditto for the above). One alternative suggestion:

Suggested change

* 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}

Updated. Thanks.

ableegoldman · 2020-11-23T19:24:27Z

                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),


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

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.

ableegoldman · 2020-11-23T19:26:00Z

-            waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, Duration.ofSeconds(15));

            TestUtils.waitForCondition(flag::get, "Handler was called");
-            assertThat(processorValueCollector.size(), equalTo(2));


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

wcarlson5

@showuon Mostly these are good changes, though @ableegoldman is right changing the default to one thread will reduce our coverage. The only reason one shutdown application uses one thread is to ensure the shutdown signal still gets propagated when the last thread is dying.

wcarlson5 · 2020-11-23T21:00:58Z

                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),


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.

wcarlson5 · 2020-11-23T21:01:11Z

-            waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, Duration.ofSeconds(15));

            TestUtils.waitForCondition(flag::get, "Handler was called");
-            assertThat(processorValueCollector.size(), equalTo(2));


wcarlson5 · 2020-11-23T21:02:18Z

-            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);


The order is not really that important here, either way works

…he test more reliable

showuon · 2020-11-24T04:33:12Z

@ableegoldman @wcarlson5 , thanks for the comments. Now I know why it sets the default treads to 2. So, to make the test more reliable, I'll do:

wait for the state become running (as before)
wait for the 1st time handler got called in current thread
wait for the 2nd time handler got called after rebalancing
wait for the stream state turned into ERROR state (as before)

This should make this test more reliable. What do you think?
commit: 2b6d0a2

wcarlson5 · 2020-11-24T16:23:16Z

These changes LGTM. WDYT @ableegoldman? thanks for the PR!

ableegoldman

Thanks for the updates, this LGTM. Looks like all tests passed this time around

ableegoldman · 2020-11-24T21:59:27Z

Merged to trunk

…t-for-generated-requests * apache-github/trunk: (405 commits) KAFKA-6687: restrict DSL to allow only Streams from the same source topics (apache#9609) MINOR: Small cleanups in `AlterIsr` handling logic (apache#9663) MINOR: Increase unit test coverage of method ProcessorTopology#updateSourceTopics() (apache#9654) MINOR: fix reading SSH output in Streams system tests (apache#9665) KAFKA-10770: Remove duplicate defination of Metrics#getTags (apache#9659) KAFKA-10722: Described the types of the used state stores (apache#9607) KAFKA-10702; Skip bookkeeping of empty transactions (apache#9632) MINOR: Remove erroneous extra <code> in design doc (apache#9657) KAFKA-10736 Convert transaction coordinator metadata schemas to use g… (apache#9611) MINOR: Update vagrant/tests readme (apache#9650) KAFKA-10720: Document prohibition on header mutation by SMTs (apache#9597) KAFKA-10713: Stricter protocol parsing in hostnames (apache#9593) KAFKA-10565: Only print console producer prompt with a tty (apache#9644) MINOR: fix listeners doc to close <code> properly (apache#9655) MINOR: Remove unnecessary statement from WorkerConnector#doRun (apache#9653) KAFKA-10758: ProcessorTopology should only consider its own nodes when updating regex source topics (apache#9648) KAFKA-10754: fix flaky tests by waiting kafka streams be in running state before assert (apache#9629) MINOR: Upgrade to Scala 2.13.4 (apache#9643) MINOR: Update build and test dependencies (apache#9645) MINOR: Remove redundant argument from TaskMetricsGroup#recordCommit (apache#9642) ... clients/src/main/java/org/apache/kafka/clients/ClientRequest.java clients/src/main/java/org/apache/kafka/clients/NetworkClient.java clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java clients/src/main/java/org/apache/kafka/common/requests/AlterConfigsResponse.java clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochResponse.java clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java clients/src/test/java/org/apache/kafka/common/requests/ProduceResponseTest.java clients/src/test/java/org/apache/kafka/common/requests/RequestContextTest.java clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java

showuon commented Nov 20, 2020

View reviewed changes

showuon force-pushed the KAFKA-10754 branch from 196891b to 0c95708 Compare November 20, 2020 14:15

showuon commented Nov 20, 2020

View reviewed changes

KAFKA-10754: fix flaky tests by waiting kafka streams be in running s…

0e1c0c1

…tate before assert

showuon force-pushed the KAFKA-10754 branch from 0c95708 to 0e1c0c1 Compare November 20, 2020 14:48

showuon commented Nov 20, 2020

View reviewed changes

mjsax added streams tests Test fixes (including flaky tests) labels Nov 20, 2020

KAFKA-10754: fix shouldShutdownThreadUsingOldHandler by using only 1 …

e6d39f6

…stream thread

showuon force-pushed the KAFKA-10754 branch from 75e2d26 to e6d39f6 Compare November 21, 2020 07:14

ableegoldman reviewed Nov 23, 2020

View reviewed changes

wcarlson5 reviewed Nov 23, 2020

View reviewed changes

KAFKA-10754: revert to set the default thread number as 2, and make t…

2b6d0a2

…he test more reliable

ableegoldman approved these changes Nov 24, 2020

View reviewed changes

ableegoldman merged commit 2a40920 into apache:trunk Nov 24, 2020

	* 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}

Conversation

showuon commented Nov 20, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon commented Nov 20, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Nov 20, 2020

Uh oh!

showuon commented Nov 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

showuon commented Nov 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon commented Nov 24, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

wcarlson5 commented Nov 24, 2020

Uh oh!

ableegoldman left a comment

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Nov 24, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

showuon commented Nov 20, 2020 •

edited

Loading

showuon commented Nov 21, 2020 •

edited

Loading

showuon commented Nov 21, 2020 •

edited

Loading

showuon commented Nov 24, 2020 •

edited

Loading