[FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with ... #10724

becketqin · 2019-12-30T14:28:32Z

…a timeout. Print stacktrace if a long closure happens.

What is the purpose of the change

Fix the test failure caused by long closing time of the the KafkaAdminClient.

Brief change log

Use the close(Duration) method in KafkaAdminClient with a timeout. It looks that this does not guarantee the timeout will be honored due to the bug in KafkaAdminClient. Therefore we also print the stacktrace of the dangling KafkaAdminClient thread in case this happens again.

Verifying this change

This change is already covered by existing tests that delete Kafka topics.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

…a timeout. Print stacktrace if a long closure happens.

flinkbot · 2019-12-30T14:32:23Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 8eaf5ae (Mon Dec 30 14:32:23 UTC 2019)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2019-12-30T15:25:24Z

CI report:

8eaf5ae Travis: SUCCESS Azure: SUCCESS
a02be8b Travis: SUCCESS Azure: SUCCESS
163dc9a Travis: FAILURE Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

carp84

+1, changes LGTM.

zentol · 2020-01-03T14:24:04Z

...afka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

 			tryDelete(adminClient, topic);
 		} catch (Exception e) {
 			e.printStackTrace();
 			fail(String.format("Delete test topic : %s failed, %s", topic, e.getMessage()));
+		} finally {
+			adminClient.close(Duration.ofMillis(30000L));


This is way too long. If our stance is that the shutdown is not relevant for correctness then let's set this to a low value (5 seconds max).

Also add a comment that we don't use a try-with-resources so that we can introduce a timeout.

zentol · 2020-01-03T14:24:10Z

...fka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java

@@ -201,4 +202,15 @@ protected NetworkFailuresProxy createProxy(String remoteHost, int remotePort) {
 		networkFailuresProxies.add(proxy);
 		return proxy;
 	}
+
+	protected void maybePrintDanglingThreadStacktrace(String threadNameKeyword) {


I'm wondering about the benefit. What do we gain by knowing where the KafkaAdminClient is blocking?

This helps to identify where exactly internal thread of KafkaAdminClient blocks. So we can address the root cause of the long closure, though it may be in Kafka instead of Flink.

Seems unlikely to ever be something in Flink though? I wouldn't want people start chasing Kafka bugs.

I think our goal is to make our tests stable, regardless of whether the bug is in Kafka or not. Adding this log is not only for debugging Kafka.

A dangling thread may not necessarily indicate a bug in Kafka. We don't know yet. For example, the blocking could be caused by our test environment setting, e.g. forget to unblock the broker traffic in some case.

Even if it is a bug in Kafka, it is better to know what exactly the bug is so we know the consequence of it. For example, in this case a topic may not be deleted, which may have impact in other tests.

carp84 · 2020-01-07T03:08:48Z

Checked and confirmed the travis failure of 163dc9a is caused by a known issue as tracked by FLINK-15247 and irrelative to changes here.

becketqin · 2020-01-07T06:10:09Z

Thanks @carp84 .

@zentol I've updated the patch. Any further comments?

becketqin · 2020-01-16T12:09:38Z

Merged.
master: fff0952
release-1.10: bf0f968

[FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with …

8eaf5ae

…a timeout. Print stacktrace if a long closure happens.

rmetzger added review=description? component=Connectors/Kafka component=Tests labels Dec 30, 2019

carp84 approved these changes Jan 2, 2020

View reviewed changes

zentol reviewed Jan 3, 2020

View reviewed changes

zentol self-assigned this Jan 3, 2020

[hotfix] Reduce the closure timeout to 5 sec.

163dc9a

becketqin force-pushed the FLINK-13595 branch from a02be8b to 163dc9a Compare January 6, 2020 15:14

zentol removed their assignment Jan 9, 2020

becketqin closed this Jan 16, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with ... #10724

[FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with ... #10724

becketqin commented Dec 30, 2019 •

edited

flinkbot commented Dec 30, 2019

flinkbot commented Dec 30, 2019 •

edited

carp84 left a comment

zentol Jan 3, 2020

zentol Jan 3, 2020

becketqin Jan 6, 2020

zentol Jan 8, 2020

becketqin Jan 9, 2020

carp84 commented Jan 7, 2020 •

edited

becketqin commented Jan 7, 2020

becketqin commented Jan 16, 2020

[FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with ... #10724

[FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with ... #10724

Conversation

becketqin commented Dec 30, 2019 • edited

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Dec 30, 2019

Automated Checks

Review Progress

flinkbot commented Dec 30, 2019 • edited

CI report:

carp84 left a comment

Choose a reason for hiding this comment

zentol Jan 3, 2020

Choose a reason for hiding this comment

zentol Jan 3, 2020

Choose a reason for hiding this comment

becketqin Jan 6, 2020

Choose a reason for hiding this comment

zentol Jan 8, 2020

Choose a reason for hiding this comment

becketqin Jan 9, 2020

Choose a reason for hiding this comment

carp84 commented Jan 7, 2020 • edited

becketqin commented Jan 7, 2020

becketqin commented Jan 16, 2020

becketqin commented Dec 30, 2019 •

edited

flinkbot commented Dec 30, 2019 •

edited

carp84 commented Jan 7, 2020 •

edited