[SPARK-4964] fix serialization issues for checkpointing

Author: koeninger

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

@@ -31,7 +31,7 @@ import kafka.consumer.{ConsumerConfig, SimpleConsumer}
   *   Requires "metadata.broker.list" or "bootstrap.servers" to be set with Kafka broker(s),
   *   NOT zookeeper servers, specified in host1:port1,host2:port2 form
   */
-class KafkaCluster(val kafkaParams: Map[String, String]) {
+class KafkaCluster(val kafkaParams: Map[String, String]) extends Serializable {
   import KafkaCluster.Err
 
   val seedBrokers: Array[(String, Int)] =
@@ -43,7 +43,15 @@ class KafkaCluster(val kafkaParams: Map[String, String]) {
         (hpa(0), hpa(1).toInt)
       }
 
-  val config: ConsumerConfig = KafkaCluster.consumerConfig(kafkaParams)
+  // ConsumerConfig isn't serializable
+  @transient private var _config: ConsumerConfig = null
+
+  def config: ConsumerConfig = this.synchronized {
+    if (_config == null) {
+      _config = KafkaCluster.consumerConfig(kafkaParams)
+    }
+    _config
+  }
 
   def connect(host: String, port: Int): SimpleConsumer =
     new SimpleConsumer(host, port, config.socketTimeoutMs,

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

@@ -64,9 +64,9 @@ class DeterministicKafkaInputDStream[
   private val kc = new KafkaCluster(kafkaParams)
 
   private val maxMessagesPerPartition: Option[Long] = {
-    val ratePerSec = ssc.sparkContext.getConf.getInt("spark.streaming.receiver.maxRate", 0)
+    val ratePerSec = context.sparkContext.getConf.getInt("spark.streaming.receiver.maxRate", 0)
     if (ratePerSec > 0) {
-      val secsPerBatch = ssc.graph.batchDuration.milliseconds.toDouble / 1000
+      val secsPerBatch = context.graph.batchDuration.milliseconds.toDouble / 1000
       Some((secsPerBatch * ratePerSec).toLong)
     } else {
       None
@@ -105,7 +105,7 @@ class DeterministicKafkaInputDStream[
   override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {
     val untilOffsets = clamp(latestLeaderOffsets(maxRetries))
     val rdd = new KafkaRDD[K, V, U, T, R](
-      ssc_.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)
+      context.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)
 
     currentOffsets = untilOffsets
     Some(rdd)