title: Kafka+Java+SparkStreaming+reduceByKeyAndWindow throw Exception:org.apache.spark.SparkException: Task not serializable
body: <p>I'm new for kafka and spark, and i'm trying to do some count, but not sucessful! The details of the problem are as follows. Thanks!</p>
  
  <p>code as belows:</p>
  
  <pre><code>JavaPairDStream&lt;String,Integer&gt; counts = wordCounts.reduceByKeyAndWindow(new AddIntegers(), new SubtractIntegers(), Durations.seconds(8000), Durations.seconds(4000));
  </code></pre>
  
  <p>Exception as belows:</p>
  
  <blockquote>
    <p>Exception in thread "Thread-3" 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:1623)     at
    org.apache.spark.streaming.dstream.PairDStreamFunctions.reduceByKeyAndWindow(PairDStreamFunctions.scala:333)
      at
    org.apache.spark.streaming.dstream.PairDStreamFunctions.reduceByKeyAndWindow(PairDStreamFunctions.scala:299)
      at
    org.apache.spark.streaming.api.java.JavaPairDStream.reduceByKeyAndWindow(JavaPairDStream.scala:352)
      at KafkaAndDstreamWithIncrement.KDDConsumer.run(KDDConsumer.java:110)
    Caused by: java.io.NotSerializableException:
    KafkaAndDstreamWithIncrement.KDDConsumer</p>
  </blockquote>