Address

Author: zsxwing

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", "100").toLong
+    sourceOptions.getOrElse("fetchOffset.retryIntervalMs", "1000").toLong
 
   private val maxOffsetsPerTrigger =
     sourceOptions.get("maxOffsetsPerTrigger").map(_.toLong)

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

@@ -814,8 +814,7 @@ class KafkaSourceStressForDontFailOnDataLossSuite extends StreamTest with Shared
 
   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")))
+    new TestSparkSession(new SparkContext("local[2,3]", "test-sql-context", sparkConf))
   }
 
   override def beforeAll(): Unit = {

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

@@ -286,8 +286,8 @@ class KafkaTestUtils extends Logging {
     props
   }
 
-  /** Assert topic is deleted in all places, e.g, brokers, zookeeper. */
-  private def assertTopicDeleted(
+  /** Verify topic is deleted in all places, e.g, brokers, zookeeper. */
+  private def verifyTopicDeletion(
       topic: String,
       numPartitions: Int,
       servers: Seq[KafkaServer]): Unit = {
@@ -320,14 +320,15 @@ class KafkaTestUtils extends Logging {
       s"topic $topic still exists on zookeeper")
   }
 
-  private def verifyTopicDeletion(
+  /** Verify topic is deleted. Retry to delete the topic if not. */
+  private def verifyTopicDeletionWithRetries(
       zkUtils: ZkUtils,
       topic: String,
       numPartitions: Int,
       servers: Seq[KafkaServer]) {
     eventually(timeout(60.seconds), interval(200.millis)) {
       try {
-        assertTopicDeleted(topic, numPartitions, servers)
+        verifyTopicDeletion(topic, numPartitions, servers)
       } catch {
         case e: Throwable =>
           // As pushing messages into Kafka updates Zookeeper asynchronously, there is a small