[SPARK-2808][Streaming][Kafka] code cleanup per TD

Author: koeninger

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaCluster.scala

@@ -38,11 +38,6 @@ private[spark]
 class KafkaCluster(val kafkaParams: Map[String, String]) extends Serializable {
   import KafkaCluster.{Err, LeaderOffset, SimpleConsumerConfig}
 
-  /** Constructor that takes a Java map */
-  def this(kafkaParams: java.util.Map[String, String]) {
-    this(kafkaParams.asScala.toMap)
-  }
-
   // ConsumerConfig isn't serializable
   @transient private var _config: SimpleConsumerConfig = null
 
@@ -227,7 +222,7 @@ class KafkaCluster(val kafkaParams: Map[String, String]) extends Serializable {
   // scalastyle:on
 
   // this 0 here indicates api version, in this case the original ZK backed api.
-  def defaultConsumerApiVersion: Short = 0
+  private def defaultConsumerApiVersion: Short = 0
 
   /** Requires Kafka >= 0.8.1.1 */
   def getConsumerOffsets(

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaTestUtils.scala

@@ -229,7 +229,7 @@ private class KafkaTestUtils extends Logging {
     tryAgain(1)
   }
 
-  /** wait until the leader offset for the given topic/partition equals the specified offset */
+  /** Wait until the leader offset for the given topic/partition equals the specified offset */
   def waitUntilLeaderOffset(
       topic: String,
       partition: Int,
@@ -245,20 +245,19 @@ private class KafkaTestUtils extends Logging {
   }
 
   private def waitUntilMetadataIsPropagated(topic: String, partition: Int): Unit = {
+    def isPropagated = server.apis.metadataCache.getPartitionInfo(topic, partition) match {
+      case Some(partitionState) =>
+        val leaderAndInSyncReplicas = partitionState.leaderIsrAndControllerEpoch.leaderAndIsr
+
+        ZkUtils.getLeaderForPartition(zkClient, topic, partition).isDefined &&
+          Request.isValidBrokerId(leaderAndInSyncReplicas.leader) &&
+          leaderAndInSyncReplicas.isr.size >= 1
+
+      case _ =>
+        false
+    }
     eventually(Time(10000), Time(100)) {
-      assert(
-        server.apis.metadataCache.getPartitionInfo(topic, partition) match {
-          case Some(partitionState) =>
-            val leaderAndIsr = partitionState.leaderIsrAndControllerEpoch.leaderAndIsr
-            ZkUtils.getLeaderForPartition(zkClient, topic, partition).isDefined &&
-              Request.isValidBrokerId(leaderAndIsr.leader) &&
-              leaderAndIsr.isr.size >= 1
-
-          case _ =>
-            false
-        },
-        s"Partition [$topic, $partition] metadata not propagated after timeout"
-      )
+      assert(isPropagated, s"Partition [$topic, $partition] metadata not propagated after timeout")
     }
   }
 

external/kafka/src/test/scala/org/apache/spark/streaming/kafka/KafkaRDDSuite.scala

@@ -101,6 +101,7 @@ class KafkaRDDSuite extends FunSuite with BeforeAndAfterAll {
 
     val rangesMap = ranges.map(o => TopicAndPartition(o.topic, o.partition) -> o.untilOffset).toMap
 
+    // make sure consumer offsets are committed before the next getRdd call
     kc.setConsumerOffsets(kafkaParams("group.id"), rangesMap).fold(
       err => throw new Exception(err.mkString("\n")),
       _ => ()

python/pyspark/streaming/tests.py

@@ -631,6 +631,7 @@ def test_kafka_direct_stream(self):
 
         self._kafkaTestUtils.createTopic(topic)
         self._kafkaTestUtils.sendMessages(topic, sendData)
+        self._kafkaTestUtils.waitUntilLeaderOffset(topic, 0, sum(sendData.values()))
 
         stream = KafkaUtils.createDirectStream(self.ssc, [topic], kafkaParams)
         self._validateStreamResult(sendData, stream)
@@ -645,6 +646,7 @@ def test_kafka_direct_stream_from_offset(self):
 
         self._kafkaTestUtils.createTopic(topic)
         self._kafkaTestUtils.sendMessages(topic, sendData)
+        self._kafkaTestUtils.waitUntilLeaderOffset(topic, 0, sum(sendData.values()))
 
         stream = KafkaUtils.createDirectStream(self.ssc, [topic], kafkaParams, fromOffsets)
         self._validateStreamResult(sendData, stream)