title: org.apache.spark.SparkException: Task not serializable
body: <p>This is a working code example:  </p>
  
  <pre><code>JavaPairDStream&lt;String, String&gt; messages = KafkaUtils.createStream(javaStreamingContext, zkQuorum, group, topicMap);
  messages.print();
  JavaDStream&lt;String&gt; lines = messages.map(new Function&lt;Tuple2&lt;String, String&gt;, String&gt;() {
      @Override
      public String call(Tuple2&lt;String, String&gt; tuple2) {
          return tuple2._2();
      }
  });
  </code></pre>
  
  <p>I get the below error:</p>
  
  <pre><code>ERROR:
  org.apache.spark.SparkException: Task not serializable
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
      at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
      at org.apache.spark.SparkContext.clean(SparkContext.scala:1435)
      at org.apache.spark.streaming.dstream.DStream.map(DStream.scala:438)
      at org.apache.spark.streaming.api.java.JavaDStreamLike$class.map(JavaDStreamLike.scala:140)
      at org.apache.spark.streaming.api.java.JavaPairDStream.map(JavaPairDStream.scala:46)
  </code></pre>