[DO NO MERGE] Debugging Kafka 0.8.2 test hang in Jenkins

external/kafka/pom.xml

@@ -44,7 +44,7 @@
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka_${scala.binary.version}</artifactId>
-      <version>0.8.1.1</version>
+      <version>0.8.2.0</version>
       <exclusions>
         <exclusion>
           <groupId>com.sun.jmx</groupId>

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

@@ -22,7 +22,7 @@ import scala.util.Random
 import scala.collection.mutable.ArrayBuffer
 import java.util.Properties
 import kafka.api._
-import kafka.common.{ErrorMapping, OffsetMetadataAndError, TopicAndPartition}
+import kafka.common.{ErrorMapping, OffsetAndMetadata, OffsetMetadataAndError, TopicAndPartition}
 import kafka.consumer.{ConsumerConfig, SimpleConsumer}
 import org.apache.spark.SparkException
 
@@ -212,12 +212,16 @@ class KafkaCluster(val kafkaParams: Map[String, String]) extends Serializable {
   // https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
   // scalastyle:on
 
+  // this 0 here indicates api version, in this case the original ZK backed api.
+  def defaultConsumerApiVersion: Short = 0
+
   /** Requires Kafka >= 0.8.1.1 */
   def getConsumerOffsets(
       groupId: String,
-      topicAndPartitions: Set[TopicAndPartition]
+      topicAndPartitions: Set[TopicAndPartition],
+      versionId: Short = defaultConsumerApiVersion
     ): Either[Err, Map[TopicAndPartition, Long]] = {
-    getConsumerOffsetMetadata(groupId, topicAndPartitions).right.map { r =>
+    getConsumerOffsetMetadata(groupId, topicAndPartitions, versionId).right.map { r =>
       r.map { kv =>
         kv._1 -> kv._2.offset
       }
@@ -227,10 +231,11 @@ class KafkaCluster(val kafkaParams: Map[String, String]) extends Serializable {
   /** Requires Kafka >= 0.8.1.1 */
   def getConsumerOffsetMetadata(
       groupId: String,
-      topicAndPartitions: Set[TopicAndPartition]
+      topicAndPartitions: Set[TopicAndPartition],
+      versionId: Short = defaultConsumerApiVersion
     ): Either[Err, Map[TopicAndPartition, OffsetMetadataAndError]] = {
     var result = Map[TopicAndPartition, OffsetMetadataAndError]()
-    val req = OffsetFetchRequest(groupId, topicAndPartitions.toSeq)
+    val req = OffsetFetchRequest(groupId, topicAndPartitions.toSeq, versionId)
     val errs = new Err
     withBrokers(Random.shuffle(config.seedBrokers), errs) { consumer =>
       val resp = consumer.fetchOffsets(req)
@@ -257,25 +262,28 @@ class KafkaCluster(val kafkaParams: Map[String, String]) extends Serializable {
   /** Requires Kafka >= 0.8.1.1 */
   def setConsumerOffsets(
       groupId: String,
-      offsets: Map[TopicAndPartition, Long]
+      offsets: Map[TopicAndPartition, Long],
+      versionId: Short = defaultConsumerApiVersion
     ): Either[Err, Map[TopicAndPartition, Short]] = {
-    setConsumerOffsetMetadata(groupId, offsets.map { kv =>
-      kv._1 -> OffsetMetadataAndError(kv._2)
-    })
+    val meta = offsets.map { kv =>
+      kv._1 -> OffsetAndMetadata(kv._2)
+    }
+    setConsumerOffsetMetadata(groupId, meta, versionId)
   }
 
   /** Requires Kafka >= 0.8.1.1 */
   def setConsumerOffsetMetadata(
       groupId: String,
-      metadata: Map[TopicAndPartition, OffsetMetadataAndError]
+      metadata: Map[TopicAndPartition, OffsetAndMetadata],
+      versionId: Short = defaultConsumerApiVersion
     ): Either[Err, Map[TopicAndPartition, Short]] = {
     var result = Map[TopicAndPartition, Short]()
-    val req = OffsetCommitRequest(groupId, metadata)
+    val req = OffsetCommitRequest(groupId, metadata, versionId)
     val errs = new Err
     val topicAndPartitions = metadata.keySet
     withBrokers(Random.shuffle(config.seedBrokers), errs) { consumer =>
       val resp = consumer.commitOffsets(req)
-      val respMap = resp.requestInfo
+      val respMap = resp.commitStatus
       val needed = topicAndPartitions.diff(result.keySet)
       needed.foreach { tp: TopicAndPartition =>
         respMap.get(tp).foreach { err: Short =>

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

@@ -85,7 +85,10 @@ class KafkaRDDSuite extends KafkaStreamSuiteBase with BeforeAndAfterAll {
     val ranges = rdd.get.asInstanceOf[HasOffsetRanges]
       .offsetRanges.map(o => TopicAndPartition(o.topic, o.partition) -> o.untilOffset).toMap
 
-    kc.setConsumerOffsets(kafkaParams("group.id"), ranges)
+    kc.setConsumerOffsets(kafkaParams("group.id"), ranges).fold(
+      err => throw new Exception(err.mkString("\n")),
+      _ => ()
+    )
 
     // this is the "0 messages" case
     val rdd2 = getRdd(kc, Set(topic))

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

@@ -88,8 +88,7 @@ abstract class KafkaStreamSuiteBase extends FunSuite with Eventually with Loggin
     var bindSuccess: Boolean = false
     while(!bindSuccess) {
       try {
-        val brokerProps = getBrokerConfig()
-        brokerConf = new KafkaConfig(brokerProps)
+        brokerConf = new KafkaConfig(brokerConfig)
         server = new KafkaServer(brokerConf)
         server.startup()
         logInfo("==================== Kafka Broker Started ====================")
@@ -137,7 +136,7 @@ abstract class KafkaStreamSuiteBase extends FunSuite with Eventually with Loggin
   def createTopic(topic: String) {
     AdminUtils.createTopic(zkClient, topic, 1, 1)
     // wait until metadata is propagated
-    waitUntilMetadataIsPropagated(topic, 0)
+    waitUntilMetadataIsPropagated(Seq(server), topic, 0)
     logInfo(s"==================== Topic $topic Created ====================")
   }
 
@@ -147,13 +146,13 @@ abstract class KafkaStreamSuiteBase extends FunSuite with Eventually with Loggin
   }
 
   def sendMessages(topic: String, messages: Array[String]) {
-    producer = new Producer[String, String](new ProducerConfig(getProducerConfig()))
+    producer = new Producer[String, String](new ProducerConfig(producerConfig))
     producer.send(messages.map { new KeyedMessage[String, String](topic, _ ) }: _*)
     producer.close()
     logInfo(s"==================== Sent Messages: ${messages.mkString(", ")} ====================")
   }
 
-  private def getBrokerConfig(): Properties = {
+  private def brokerConfig: Properties = {
     val props = new Properties()
     props.put("broker.id", "0")
     props.put("host.name", "localhost")
@@ -165,21 +164,28 @@ abstract class KafkaStreamSuiteBase extends FunSuite with Eventually with Loggin
     props
   }
 
-  private def getProducerConfig(): Properties = {
+  private def producerConfig: Properties = {
     val brokerAddr = brokerConf.hostName + ":" + brokerConf.port
     val props = new Properties()
     props.put("metadata.broker.list", brokerAddr)
     props.put("serializer.class", classOf[StringEncoder].getName)
     props
   }
 
-  private def waitUntilMetadataIsPropagated(topic: String, partition: Int) {
-    eventually(timeout(10000 milliseconds), interval(100 milliseconds)) {
-      assert(
-        server.apis.metadataCache.containsTopicAndPartition(topic, partition),
-        s"Partition [$topic, $partition] metadata not propagated after timeout"
-      )
+  private def waitUntilMetadataIsPropagated(servers: Seq[KafkaServer], topic: String, partition: Int): Int = {
+    var leader: Int = -1
+    eventually(timeout(1000 milliseconds), interval(100 milliseconds)) {
+      assert(servers.forall { server =>
+        val partitionStateOpt = server.apis.metadataCache.getPartitionInfo(topic, partition)
+        partitionStateOpt match {
+          case Some(partitionState) =>
+            leader = partitionState.leaderIsrAndControllerEpoch.leaderAndIsr.leader
+            leader >= 0 // is valid broker id
+          case _ => false
+        }
+      }, s"Partition [$topic, $partition] metadata not propagated after timeout")
     }
+    leader
   }
 
   class EmbeddedZookeeper(val zkConnect: String) {