[SPARK-25116][TESTS]Fix the Kafka cluster leak and clean up cached producers #22106

zsxwing · 2018-08-14T18:12:58Z

What changes were proposed in this pull request?

KafkaContinuousSinkSuite leaks a Kafka cluster because both KafkaSourceTest and KafkaContinuousSinkSuite create a Kafka cluster but afterAll only shuts down one cluster. This leaks a Kafka cluster and causes that some Kafka thread crash and kill JVM when SBT is trying to clean up tests.

This PR fixes the leak and also adds a shut down hook to detect Kafka cluster leak.

In additions, it also fixes AdminClient leak and cleans up cached producers (When a record is writtn using a producer, the producer will keep refreshing the topic and I don't find an API to clear it except closing the producer) to eliminate the following annoying logs:

8/13 15:34:42.568 kafka-admin-client-thread | adminclient-4 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node 0 could not be established. Broker may not be available.
18/08/13 15:34:42.570 kafka-admin-client-thread | adminclient-6 WARN NetworkClient: [AdminClient clientId=adminclient-6] Connection to node 0 could not be established. Broker may not be available.
18/08/13 15:34:42.606 kafka-admin-client-thread | adminclient-8 WARN NetworkClient: [AdminClient clientId=adminclient-8] Connection to node -1 could not be established. Broker may not be available.
18/08/13 15:34:42.729 kafka-producer-network-thread | producer-797 WARN NetworkClient: [Producer clientId=producer-797] Connection to node -1 could not be established. Broker may not be available.
18/08/13 15:34:42.906 kafka-producer-network-thread | producer-1598 WARN NetworkClient: [Producer clientId=producer-1598] Connection to node 0 could not be established. Broker may not be available.

I also reverted b5eb542 introduced by #22097 since it doesn't help.

How was this patch tested?

Jenkins

This reverts commit b5eb542.

SparkQA · 2018-08-14T18:19:12Z

Test build #4248 has finished for PR 22106 at commit 574f5fa.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T18:21:02Z

Test build #94761 has finished for PR 22106 at commit 574f5fa.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T19:02:59Z

Test build #4249 has finished for PR 22106 at commit 57a52a9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T19:03:26Z

Test build #4250 has finished for PR 22106 at commit 57a52a9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T19:10:34Z

Test build #4251 has finished for PR 22106 at commit 57a52a9.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T19:15:37Z

Test build #4252 has finished for PR 22106 at commit 57a52a9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T19:17:04Z

Test build #94763 has finished for PR 22106 at commit 57a52a9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T23:45:14Z

Test build #94774 has finished for PR 22106 at commit f01cf8d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T23:47:30Z

Test build #4253 has finished for PR 22106 at commit f01cf8d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T00:07:34Z

Test build #4255 has finished for PR 22106 at commit f01cf8d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T00:07:39Z

Test build #4256 has finished for PR 22106 at commit f01cf8d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T00:08:22Z

Test build #4254 has finished for PR 22106 at commit f01cf8d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T00:14:58Z

Test build #4257 has finished for PR 22106 at commit f01cf8d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2018-08-15T19:38:02Z

...rnal/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaContinuousReader.scala

@@ -216,7 +216,7 @@ class KafkaContinuousInputPartitionReader(
      } catch {
        // We didn't read within the timeout. We're supposed to block indefinitely for new data, so
        // swallow and ignore this.
-        case _: TimeoutException =>
+        case _: TimeoutException | _: org.apache.kafka.common.errors.TimeoutException =>


This is to fix https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4254/

org.apache.kafka.common.errors.TimeoutException: Timeout of 3000ms expired before the position for partition failOnDataLoss-2-0 could be determined triggered a task retry but as continuous processing doesn't support task retries, it failed with org.apache.spark.sql.execution.streaming.continuous.ContinuousTaskRetryException: Continuous execution does not support task retry.

zsxwing · 2018-08-15T19:50:17Z

...l/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousSinkSuite.scala

-      withBrokerProps = Map("auto.create.topics.enable" -> "false"))
-    testUtils.setup()
-  }
+  override val brokerProps = Map("auto.create.topics.enable" -> "false")


This is the fix for Kafka cluster leak

SparkQA · 2018-08-15T20:21:03Z

Test build #94815 has finished for PR 22106 at commit 621099e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T20:26:48Z

Test build #4262 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T20:26:58Z

Test build #94816 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T20:28:57Z

Test build #4259 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T20:29:09Z

Test build #4260 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T20:31:05Z

Test build #4261 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-15T21:43:01Z

Test build #4265 has started for PR 22106 at commit 63cc11d.

SparkQA · 2018-08-15T21:43:19Z

Test build #4267 has started for PR 22106 at commit 63cc11d.

SparkQA · 2018-08-15T22:18:48Z

Test build #4268 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:29:46Z

Test build #4273 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:33:37Z

Test build #4275 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:34:04Z

Test build #4270 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:35:36Z

Test build #4274 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:43:18Z

Test build #4269 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:43:41Z

Test build #4271 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:44:34Z

Test build #4272 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T00:45:12Z

Test build #4276 has finished for PR 22106 at commit 63cc11d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T01:55:10Z

Test build #4266 has finished for PR 22106 at commit 63cc11d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-16T02:00:25Z

Test build #4264 has finished for PR 22106 at commit 63cc11d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2018-08-17T16:35:56Z

Test failures in 4264 and 4266 are unrelated. The latest changes passed on Jenkins 15 times.

zsxwing · 2018-08-17T16:36:07Z

cc @srowen

zsxwing · 2018-08-17T17:24:04Z

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/CachedKafkaProducer.scala

  private lazy val cacheExpireTimeout: Long =
-    SparkEnv.get.conf.getTimeAsMs("spark.kafka.producer.cache.timeout", "10m")
+    Option(SparkEnv.get).map(_.conf.getTimeAsMs(


Change this to call clear in afterAll even if the SparkContext has been stopped.

brkyvz · 2018-08-17T20:57:25Z

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala

@@ -130,6 +130,13 @@ class KafkaTestUtils(withBrokerProps: Map[String, Object] = Map.empty) extends L

  /** setup the whole embedded servers, including Zookeeper and Kafka brokers */
  def setup(): Unit = {
+    // Set up a KafkaTestUtils leak detector so that we can see where the leak KafkaTestUtils is
+    // created.
+    val exception = new SparkException("It was created at: ")


brkyvz · 2018-08-17T20:59:16Z

external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala

-        producer.close()
-        producer = null
-      }
+    brokerReady = false


do these need to be thread safe? Is boot up, and cleanup serial?

No. We set up in beforeAll and clean up in afterAll, which will be in the same thread.

brkyvz

LGTM. Had one question

zsxwing · 2018-08-17T21:20:20Z

Thanks. Merging to master.

zsxwing added 5 commits August 14, 2018 10:59

fix kafka cluster leak

6a0ec9c

Revert "don't kill JVM during termination"

59d10e2

This reverts commit b5eb542.

Merge branch 'master' into SPARK-25116

e13b21a

update

b561528

update

574f5fa

zsxwing changed the title ~~[SPARK-25116][Tests]Fix the kafka cluster leak and clean up cached producers~~ [SPARK-25116][TESTS]Fix the Kafka cluster leak and clean up cached producers Aug 14, 2018

style

57a52a9

increase zookeeper session timeout

f01cf8d

fix more

621099e

zsxwing commented Aug 15, 2018

View reviewed changes

more

63cc11d

zsxwing commented Aug 15, 2018

View reviewed changes

zsxwing commented Aug 17, 2018

View reviewed changes

brkyvz reviewed Aug 17, 2018

View reviewed changes

brkyvz approved these changes Aug 17, 2018

View reviewed changes

asfgit closed this in da2dc69 Aug 17, 2018

zsxwing deleted the SPARK-25116 branch August 17, 2018 21:29

attilapiros mentioned this pull request Apr 3, 2019

[SPARK-27343][KAFKA][SS]Avoid hardcoded for spark-sql-kafka-0-10 #24270

Closed

[SPARK-25116][TESTS]Fix the Kafka cluster leak and clean up cached producers #22106

[SPARK-25116][TESTS]Fix the Kafka cluster leak and clean up cached producers #22106

Conversation

zsxwing commented Aug 14, 2018 • edited

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

zsxwing Aug 15, 2018

Choose a reason for hiding this comment

zsxwing Aug 15, 2018

Choose a reason for hiding this comment

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

SparkQA commented Aug 16, 2018

zsxwing commented Aug 17, 2018

zsxwing commented Aug 17, 2018

zsxwing Aug 17, 2018

Choose a reason for hiding this comment

brkyvz Aug 17, 2018

Choose a reason for hiding this comment

brkyvz Aug 17, 2018

Choose a reason for hiding this comment

zsxwing Aug 17, 2018

Choose a reason for hiding this comment

brkyvz left a comment

Choose a reason for hiding this comment

zsxwing commented Aug 17, 2018

zsxwing commented Aug 14, 2018 •

edited