2.2.1 build fixed (apache#231)

* MapR [SPARK-164] Update Kafka version to 1.0.1-mapr in Spark Kafka Producer module

Author: Mikhail Gorbov

examples/src/main/scala/org/apache/spark/examples/streaming/KafkaProducerExample.scala

@@ -79,11 +79,11 @@ object KafkaProducerExample extends App {
   val producerConf = new ProducerConf(
     bootstrapServers = kafkaBrokers.split(",").toList)
 
-  val items = (0 until numMessages.toInt).map(i => Item(i, i))
-  val defaultRDD: RDD[Item] = ssc.sparkContext.parallelize(items)
-  val dStream: DStream[Item] = new ConstantInputDStream[Item](ssc, defaultRDD)
+  val items = (0 until numMessages.toInt).map(i => Item(i, i).toString)
+  val defaultRDD: RDD[String] = ssc.sparkContext.parallelize(items)
+  val dStream: DStream[String] = new ConstantInputDStream[String](ssc, defaultRDD)
 
-  dStream.sendToKafka[ItemJsonSerializer](topics, producerConf)
+  dStream.foreachRDD(_.sendToKafka(topics, producerConf))
   dStream.count().print()
 
   ssc.start()

external/kafka-producer/pom.xml

@@ -51,7 +51,6 @@
             <groupId>org.eclipse.jetty</groupId>
             <artifactId>jetty-util</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-streaming_${scala.binary.version}</artifactId>
@@ -120,6 +119,11 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-library</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>net.sf.jopt-simple</groupId>
             <artifactId>jopt-simple</artifactId>

external/kafka-producer/src/main/scala/org/apache/spark/streaming/kafka/producer/KafkaRDDWriter.scala

@@ -33,17 +33,14 @@ case class ProducerConf(
     linger: Long = 0,
     others: Map[String, String] = Map.empty[String, String]) {
 
-  private var keySerializer: Option[String] = null
-
-  private var valueSerializer: Option[String] = null
+  private var keySerializer: String = ProducerConf.ByteArraySerializer
+  private var valueSerializer: String = ProducerConf.StringSerializer
 
   def asJMap(): java.util.Map[String, Object] = {
     val kafkaParams = Map[String, AnyRef](
       ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers.mkString(","),
-      ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG -> keySerializer.getOrElse(
-        ProducerConf.ByteArraySerializer),
-      ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG -> valueSerializer.getOrElse(
-        ProducerConf.StringSerializer),
+      ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG -> keySerializer,
+      ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG -> valueSerializer,
       ProducerConfig.ACKS_CONFIG -> acks,
       ProducerConfig.BUFFER_MEMORY_CONFIG -> bufferMemory.toString,
       ProducerConfig.COMPRESSION_TYPE_CONFIG -> compressionType.getOrElse("none"),
@@ -56,12 +53,12 @@ case class ProducerConf(
   }
 
   def withKeySerializer(serializer: String): ProducerConf = {
-    this.keySerializer = Some(serializer)
+    this.keySerializer = serializer
     this
   }
 
   def withValueSerializer(serializer: String): ProducerConf = {
-    this.valueSerializer = Some(serializer)
+    this.valueSerializer = serializer
     this
   }
 }
@@ -71,7 +68,6 @@ object ProducerConf {
   private val Prefix = "spark.streaming.kafka"
 
   val ByteArraySerializer = "org.apache.kafka.common.serialization.ByteArraySerializer"
-
   val StringSerializer = "org.apache.kafka.common.serialization.StringSerializer"
 
   def apply(sparkConf: SparkConf): ProducerConf = {

external/kafka-producer/src/main/scala/org/apache/spark/streaming/kafka/producer/ProducerConf.scala

@@ -17,29 +17,30 @@
 
 package org.apache.spark.streaming.kafka.producer
 
-import scala.reflect.ClassTag
+import scala.language.implicitConversions
 
-import org.apache.kafka.common.serialization.{ByteArraySerializer, Serializer}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
 
 import org.apache.spark.rdd.RDD
 
-class RDDFunctions[T: ClassTag](rdd: RDD[T]) extends Serializable {
-  def sendToKafka[S <: Serializer[T] : ClassTag](
-    topic: String,
-    conf: ProducerConf): Unit = {
-    rdd.sparkContext.runJob(
-      rdd,
-      new KafkaRDDWriter[Array[Byte], T, ByteArraySerializer, S](topic, conf).sendV _)
+class RDDFunctions[T](rdd: RDD[T]) {
+  def sendToKafka(topic: String, conf: ProducerConf): Unit = {
+    rdd.foreachPartition(iter => {
+      val producer = new KafkaProducer[String, T](conf.asJMap())
+      iter.foreach { item =>
+        producer.send(new ProducerRecord[String, T](topic, item))
+      }
+    })
   }
 }
 
-class PairRDDFunctions[K: ClassTag, V: ClassTag](rdd: RDD[(K, V)]) extends Serializable {
-  def sendToKafka[
-    KS <: Serializer[K] : ClassTag,
-    VS <: Serializer[V] : ClassTag
-  ](topic: String, conf: ProducerConf): Unit = {
-    rdd.sparkContext.runJob(
-      rdd,
-      new KafkaRDDWriter[K, V, KS, VS](topic, conf).sendKV _)
+class PairRDDFunctions[K, V](rdd: RDD[(K, V)]) {
+  def sendToKafka(topic: String, conf: ProducerConf): Unit = {
+    rdd.foreachPartition(iter => {
+      val producer = new KafkaProducer[K, V](conf.asJMap())
+      iter.foreach { item =>
+        producer.send(new ProducerRecord[K, V](topic, item._1, item._2))
+      }
+    })
   }
 }

external/kafka-producer/src/main/scala/org/apache/spark/streaming/kafka/producer/RDDFunctions.scala

@@ -17,16 +17,17 @@
 
 package org.apache.spark.streaming.kafka.producer
 
+import scala.collection.JavaConverters._
+import scala.concurrent.duration._
+import scala.language.postfixOps
+
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.serialization.StringSerializer
+
 import org.apache.spark.rdd.RDD
 import org.apache.spark.{SparkConf, SparkContext}
 
-import scala.collection.JavaConverters._
-import scala.concurrent.duration._
-import scala.language.postfixOps
-
 class RDDProducerSuite extends BaseKafkaProducerTest {
 
   private val numMessages = 10
@@ -51,7 +52,7 @@ class RDDProducerSuite extends BaseKafkaProducerTest {
 
     val rdd: RDD[String] = sparkContext.parallelize(List.fill(numMessages)(recordValue))
 
-    rdd.sendToKafka[StringSerializer](topic, testConf)
+    rdd.sendToKafka(topic, testConf)
 
     consumer.assign(List(new TopicPartition(topic, partition)).asJava)
 
@@ -75,7 +76,7 @@ class RDDProducerSuite extends BaseKafkaProducerTest {
 
     val rdd = sparkContext.parallelize(List.fill(numMessages)(recordKey, recordValue))
 
-    rdd.sendToKafka[StringSerializer, StringSerializer](topic, testConf)
+    rdd.sendToKafka(topic, testConf)
 
     consumer.assign(List(new TopicPartition(topic, partition)).asJava)