title: Kafka Spark Producer Receiver Communication
body: <p>I am trying to create a basic Kafka producer and receiver just to understand how things work. The code below doesn't yield any errors, but my receiver doesn't seem to be receiving anything. Also, the producer seems to be sending very slowly. </p>
  
  <p>Can anyone tell me why the receiver isn't printing anything, and why the producer is sending so slow?</p>
  
  <p>Here's my producer that I run first:</p>
  
  <pre><code>import org.apache.http.util.EntityUtils
  import java.util.Properties
  import org.apache.kafka.common.serialization.StringSerializer
  import org.apache.kafka.clients.producer.{KafkaProducer,ProducerRecord}
  
  object ProducerTest{
  
  
    def extractOptions(properties: Map[String, Any]): Properties = {
      val props = new Properties()
      properties.foreach { case (key, value) =&gt; props.put(key, value.toString)     }
      props
    }
  
  
    val mandatoryOptions: Map[String, Any] = Map(
      "bootstrap.servers" -&gt; "127.0.0.1:9092",
      "acks" -&gt; "all",
      "batch.size" -&gt; 16384,
      "linger.ms" -&gt; 1,
      "buffer.memory" -&gt; 33554432,
      "key.serializer" -&gt;     "org.apache.kafka.common.serialization.StringSerializer",
      "value.serializer" -&gt;     "org.apache.kafka.common.serialization.StringSerializer")
  
    val producer = new KafkaProducer[String, String]    (extractOptions(mandatoryOptions))
  
    val TOPIC="test"
  
    def urlRecall {
  
      println("Hallo")
  
      val record = new ProducerRecord(TOPIC, "key", "hallo")
      producer.send(record,null)
  
      urlRecall
  
    }
  
    def main(args: Array[String]) {
  
      urlRecall
  
    }
  
    ;
  }
  </code></pre>
  
  <p>The code below shows the receiver that I run in a separate window. It doesn't give any errors, but doesn't print anything either:</p>
  
  <pre><code>import org.apache.kafka.common.serialization.StringDeserializer
  import org.apache.spark.streaming.kafka010._
  import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
  import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
  import org.apache.spark._
  import org.apache.spark.streaming._
  
  object kafkaConsumerTest extends App {
  
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -&gt; "127.0.0.1:9092",
      "key.deserializer" -&gt; classOf[StringDeserializer],
      "value.deserializer" -&gt; classOf[StringDeserializer],
      "group.id" -&gt; "1",
      "auto.offset.reset" -&gt; "latest",
      "enable.auto.commit" -&gt; (false: java.lang.Boolean)
    )
  
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(conf, Seconds(1))
  
    val topics = Array("test")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
  
    stream.print()
  
    ssc.start()
    ssc.awaitTermination()
  
  }
  </code></pre>