[flink] branch release-1.10 updated: [FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with a timeout. Print stacktrace if a long closure happens.

[jq...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jqin pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.10 by this push:
     new bf0f968  [FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with a timeout. Print stacktrace if a long closure happens.
bf0f968 is described below

commit bf0f968f89061153d0d3a93890eb8f7a202f0c59
Author: Jiangjie (Becket) Qin <ji...@alibaba-inc.com>
AuthorDate: Wed Dec 25 14:16:52 2019 +0800

    [FLINK-13595][connector/kafka][test] Close the KafkaAdminClient with a timeout. Print stacktrace if a long closure happens.
---
 .../streaming/connectors/kafka/KafkaTestEnvironment.java    | 12 ++++++++++++
 .../connectors/kafka/KafkaTestEnvironmentImpl.java          | 13 ++++++++++++-
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
index 3df8dbc..37132e7 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
@@ -33,6 +33,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 
 /**
@@ -209,4 +210,15 @@ public abstract class KafkaTestEnvironment {
 		networkFailuresProxies.add(proxy);
 		return proxy;
 	}
+
+	protected void maybePrintDanglingThreadStacktrace(String threadNameKeyword) {
+		for (Map.Entry<Thread, StackTraceElement[]> threadEntry : Thread.getAllStackTraces().entrySet()) {
+			if (threadEntry.getKey().getName().contains(threadNameKeyword)) {
+				System.out.println("Dangling thread found:");
+				for (StackTraceElement ste : threadEntry.getValue()) {
+					System.out.println(ste);
+				}
+			}
+		}
+	}
 }
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index f2c6f96..15b1594 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -47,6 +47,7 @@ import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.net.BindException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -57,6 +58,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -148,11 +150,20 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
 	@Override
 	public void deleteTestTopic(String topic) {
 		LOG.info("Deleting topic {}", topic);
-		try (AdminClient adminClient = AdminClient.create(getStandardProperties())) {
+		Properties props = getSecureProperties();
+		props.putAll(getStandardProperties());
+		String clientId = Long.toString(new Random().nextLong());
+		props.put("client.id", clientId);
+		AdminClient adminClient = AdminClient.create(props);
+		// We do not use a try-catch clause here so we can apply a timeout to the admin client closure.
+		try {
 			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(5000L));
+			maybePrintDanglingThreadStacktrace(clientId);
 		}
 	}