[SPARK-18588][Tests]Fix flaky test: KafkaSourceStressForDontFailOnDataLossSuite #16109

zsxwing · 2016-12-01T21:47:34Z

What changes were proposed in this pull request?

Fixed the following failures:

org.scalatest.exceptions.TestFailedDueToTimeoutException: The code passed to eventually never returned normally. Attempted 3745 times over 1.0000790851666665 minutes. Last failure message: assertion failed: failOnDataLoss-0 not deleted after timeout.

sbt.ForkMain$ForkError: org.apache.spark.sql.streaming.StreamingQueryException: Query query-66 terminated with exception: null
	at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:252)
	at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:146)
Caused by: sbt.ForkMain$ForkError: java.lang.NullPointerException: null
	at java.util.ArrayList.addAll(ArrayList.java:577)
	at org.apache.kafka.clients.Metadata.getClusterForCurrentTopics(Metadata.java:257)
	at org.apache.kafka.clients.Metadata.update(Metadata.java:177)
	at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleResponse(NetworkClient.java:605)
	at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeHandleCompletedReceive(NetworkClient.java:582)
	at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:450)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:269)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitPendingRequests(ConsumerNetworkClient.java:260)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:222)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.ensurePartitionAssignment(ConsumerCoordinator.java:366)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:978)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:938)
	at 
...

How was this patch tested?

Tested in #16048 by running many times.

zsxwing · 2016-12-01T21:48:24Z

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

@@ -102,7 +102,7 @@ private[kafka010] case class KafkaSource(
    sourceOptions.getOrElse("fetchOffset.numRetries", "3").toInt

  private val offsetFetchAttemptIntervalMs =
-    sourceOptions.getOrElse("fetchOffset.retryIntervalMs", "10").toLong
+    sourceOptions.getOrElse("fetchOffset.retryIntervalMs", "100").toLong


Increase this value because if not sleeping enough time, it's easy to fail with the same error when deleting a topic.

Is 100ms enough or should this be increase to 1 second?

Increased the default value to 1 second.

zsxwing · 2016-12-01T21:49:07Z

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

  private def verifyTopicDeletion(
      zkUtils: ZkUtils,
      topic: String,
      numPartitions: Int,
      servers: Seq[KafkaServer]) {
-    import ZkUtils._
-    val topicAndPartitions = (0 until numPartitions).map(TopicAndPartition(topic, _))
-    def isDeleted(): Boolean = {


Renamed this method to assertTopicDeleted and moved out of this method. I also rewrote it to make it tell us which line fails.

zsxwing · 2016-12-01T21:49:44Z

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

@@ -331,7 +351,7 @@ class KafkaTestUtils extends Logging {
      case _ =>
        false
    }
-    eventually(timeout(10.seconds)) {
+    eventually(timeout(60.seconds)) {


Increase this timeout as it seems 10 seconds is not enough sometimes.

SparkQA · 2016-12-01T21:51:41Z

Test build #3451 has started for PR 16109 at commit 0901a65.

SparkQA · 2016-12-01T21:52:04Z

Test build #3452 has started for PR 16109 at commit 0901a65.

SparkQA · 2016-12-02T00:14:39Z

Test build #3453 has finished for PR 16109 at commit 0901a65.

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

SparkQA · 2016-12-02T00:17:07Z

Test build #3450 has finished for PR 16109 at commit 0901a65.

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

SparkQA · 2016-12-02T01:03:22Z

Test build #69508 has finished for PR 16109 at commit 0901a65.

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

SparkQA · 2016-12-02T21:01:43Z

Test build #3459 has finished for PR 16109 at commit 0901a65.

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

SparkQA · 2016-12-02T21:23:10Z

Test build #3457 has finished for PR 16109 at commit 0901a65.

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

SparkQA · 2016-12-02T21:31:33Z

Test build #3456 has finished for PR 16109 at commit 0901a65.

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

SparkQA · 2016-12-02T21:32:31Z

Test build #3458 has finished for PR 16109 at commit 0901a65.

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

SparkQA · 2016-12-02T21:33:51Z

Test build #3460 has finished for PR 16109 at commit 0901a65.

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

zsxwing · 2016-12-03T00:40:05Z

Test build #3459 has finished for PR 16109 at commit 0901a65.

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

Not related to this PR.

Test build #3450 has finished for PR 16109 at commit 0901a65.

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

Not related to this PR.

Test build #3453 has finished for PR 16109 at commit 0901a65.

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

Looks like the retry interval is no enough. Found the following logs:

16/12/01 16:14:27.199 stream execution thread for query-62 WARN KafkaSource: Error in attempt 1 getting Kafka offsets: 
java.lang.NullPointerException
16/12/01 16:14:30.832 stream execution thread for query-62 WARN KafkaSource: Error in attempt 2 getting Kafka offsets: 
java.lang.NullPointerException
16/12/01 16:14:34.271 stream execution thread for query-62 WARN KafkaSource: Error in attempt 3 getting Kafka offsets:
java.lang.NullPointerException

I will increase the interval.

SparkQA · 2016-12-03T03:03:20Z

Test build #3461 has finished for PR 16109 at commit 8c5ae2d.

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

SparkQA · 2016-12-03T03:32:42Z

Test build #69598 has finished for PR 16109 at commit 8c5ae2d.

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

SparkQA · 2016-12-03T03:34:27Z

Test build #3463 has finished for PR 16109 at commit 8c5ae2d.

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

SparkQA · 2016-12-03T03:36:49Z

Test build #3464 has finished for PR 16109 at commit 8c5ae2d.

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

SparkQA · 2016-12-03T03:37:48Z

Test build #3462 has finished for PR 16109 at commit 8c5ae2d.

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

zsxwing · 2016-12-05T19:00:05Z

The NPE without stack trace is probably because http://jawspeak.com/2010/05/26/hotspot-caused-exceptions-to-lose-their-stack-traces-in-production-and-the-fix/

zsxwing · 2016-12-06T21:31:46Z

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

+          }
+        }
+      } else if (record.offset < offset) {
+        // This should not happen. If it does happen, then we probably misunderstand Kafka internal


Based on https://github.com/apache/kafka/blob/0.10.0.1/core/src/main/scala/kafka/log/Log.scala#L508

zsxwing · 2016-12-06T21:32:03Z

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

-      assert(record.offset == offset,
-        s"The fetched data has a different offset: expected $offset but was ${record.offset}")
-      record
+      // In general, Kafka uses the specified offset as the start point, and tries to fetch the next


Based on https://github.com/apache/kafka/blob/0.10.0.1/core/src/main/scala/kafka/log/Log.scala#L488

SparkQA · 2016-12-06T23:38:44Z

Test build #69746 has finished for PR 16109 at commit 20b01d4.

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

zsxwing · 2016-12-06T23:46:26Z

retest this please

tdas · 2016-12-07T01:27:20Z

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

+      !zkUtils.getAllTopics().contains(topic),
+      s"topic $topic still exists on zookeeper")
+  }
+
  private def verifyTopicDeletion(


its hard to differentiate the semantics of verifyTopicDeletion and assertTopicDeleted. How about renamed them to verifyTopicDeletionWithRetries and verifyTopicDeletion?

tdas · 2016-12-07T01:28:39Z

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

+  override def createSparkSession(): TestSparkSession = {
+    // Set maxRetries to 3 to handle NPE from `poll` when deleting a topic
+    new TestSparkSession(new SparkContext("local[2,3]", "test-sql-context",
+      sparkConf.set("spark.sql.testkey", "true")))


what is this key for?

It's only used on SQLConfSuite. Removed it from this test.

SparkQA · 2016-12-07T02:17:49Z

Test build #69752 has finished for PR 16109 at commit 20b01d4.

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

tdas · 2016-12-07T03:51:35Z

overall LGTM, just a few nits that would be good to fix.

SparkQA · 2016-12-07T08:13:52Z

Test build #69784 has finished for PR 16109 at commit c9d5ba9.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-12-07T10:49:45Z

Test build #69786 has finished for PR 16109 at commit c70ff19.

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

tdas · 2016-12-07T21:39:02Z

LGTM. Merging this to master and 2.1

…taLossSuite ## What changes were proposed in this pull request? Fixed the following failures: ``` org.scalatest.exceptions.TestFailedDueToTimeoutException: The code passed to eventually never returned normally. Attempted 3745 times over 1.0000790851666665 minutes. Last failure message: assertion failed: failOnDataLoss-0 not deleted after timeout. ``` ``` sbt.ForkMain$ForkError: org.apache.spark.sql.streaming.StreamingQueryException: Query query-66 terminated with exception: null at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:252) at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:146) Caused by: sbt.ForkMain$ForkError: java.lang.NullPointerException: null at java.util.ArrayList.addAll(ArrayList.java:577) at org.apache.kafka.clients.Metadata.getClusterForCurrentTopics(Metadata.java:257) at org.apache.kafka.clients.Metadata.update(Metadata.java:177) at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleResponse(NetworkClient.java:605) at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeHandleCompletedReceive(NetworkClient.java:582) at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:450) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:269) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitPendingRequests(ConsumerNetworkClient.java:260) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:222) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.ensurePartitionAssignment(ConsumerCoordinator.java:366) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:978) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:938) at ... ``` ## How was this patch tested? Tested in #16048 by running many times. Author: Shixiong Zhu <shixiong@databricks.com> Closes #16109 from zsxwing/fix-kafka-flaky-test. (cherry picked from commit edc87e1) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

…taLossSuite ## What changes were proposed in this pull request? Fixed the following failures: ``` org.scalatest.exceptions.TestFailedDueToTimeoutException: The code passed to eventually never returned normally. Attempted 3745 times over 1.0000790851666665 minutes. Last failure message: assertion failed: failOnDataLoss-0 not deleted after timeout. ``` ``` sbt.ForkMain$ForkError: org.apache.spark.sql.streaming.StreamingQueryException: Query query-66 terminated with exception: null at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:252) at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:146) Caused by: sbt.ForkMain$ForkError: java.lang.NullPointerException: null at java.util.ArrayList.addAll(ArrayList.java:577) at org.apache.kafka.clients.Metadata.getClusterForCurrentTopics(Metadata.java:257) at org.apache.kafka.clients.Metadata.update(Metadata.java:177) at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleResponse(NetworkClient.java:605) at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeHandleCompletedReceive(NetworkClient.java:582) at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:450) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:269) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitPendingRequests(ConsumerNetworkClient.java:260) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:222) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.ensurePartitionAssignment(ConsumerCoordinator.java:366) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:978) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:938) at ... ``` ## How was this patch tested? Tested in apache#16048 by running many times. Author: Shixiong Zhu <shixiong@databricks.com> Closes apache#16109 from zsxwing/fix-kafka-flaky-test.

Fix flaky test: KafkaSourceStressForDontFailOnDataLossSuite

0901a65

zsxwing commented Dec 1, 2016

View reviewed changes

Increase fetchOffset.retryIntervalMs to 3 seconds

8c5ae2d

Handle offset mismatch

20b01d4

zsxwing commented Dec 6, 2016

View reviewed changes

tdas reviewed Dec 7, 2016

View reviewed changes

zsxwing added 2 commits December 7, 2016 00:04

Address

c9d5ba9

Fix

c70ff19

asfgit closed this in edc87e1 Dec 7, 2016

zsxwing deleted the fix-kafka-flaky-test branch December 7, 2016 23:29

[SPARK-18588][Tests]Fix flaky test: KafkaSourceStressForDontFailOnDataLossSuite #16109

[SPARK-18588][Tests]Fix flaky test: KafkaSourceStressForDontFailOnDataLossSuite #16109

Conversation

zsxwing commented Dec 1, 2016

What changes were proposed in this pull request?

How was this patch tested?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zsxwing Dec 1, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 1, 2016

SparkQA commented Dec 1, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

zsxwing commented Dec 3, 2016

SparkQA commented Dec 3, 2016

SparkQA commented Dec 3, 2016

SparkQA commented Dec 3, 2016

SparkQA commented Dec 3, 2016

SparkQA commented Dec 3, 2016

zsxwing commented Dec 5, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 6, 2016

zsxwing commented Dec 6, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 7, 2016

tdas commented Dec 7, 2016

SparkQA commented Dec 7, 2016

SparkQA commented Dec 7, 2016

tdas commented Dec 7, 2016

zsxwing Dec 1, 2016 •

edited

Loading