Fix test

Andrew Or · Andrew Or · commit b3806aba7298 · 2015-05-14T12:29:33.000-07:00
diff --git a/external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala b/external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala
@@ -233,7 +233,8 @@ object KafkaUtils {
         case (tp: TopicAndPartition, Broker(host, port)) => (tp, (host, port))
       }.toMap
     }
-    new KafkaRDD[K, V, KD, VD, R](sc, kafkaParams, offsetRanges, leaderMap, messageHandler)
+    val cleanedHandler = sc.clean(messageHandler)
+    new KafkaRDD[K, V, KD, VD, R](sc, kafkaParams, offsetRanges, leaderMap, cleanedHandler)
   }
 
   /**

Original file line number	Diff line number	Diff line change
`@@ -233,7 +233,8 @@ object KafkaUtils {`
`233`	`233`	`case (tp: TopicAndPartition, Broker(host, port)) => (tp, (host, port))`
`234`	`234`	`}.toMap`
`235`	`235`	`}`
`236`		`- new KafkaRDD[K, V, KD, VD, R](sc, kafkaParams, offsetRanges, leaderMap, messageHandler)`
	`236`	`+ val cleanedHandler = sc.clean(messageHandler)`
	`237`	`+ new KafkaRDD[K, V, KD, VD, R](sc, kafkaParams, offsetRanges, leaderMap, cleanedHandler)`
`237`	`238`	`}`
`238`	`239`
`239`	`240`	`/**`