title: Avro serialization object not serializable issue
body: <p>I am trying to produce and consume Avro messages using Kafka through spark streaming API. But Avro throws object not the serializable exception. I tried wrapping the datum using an AvroKey wrapper. Still, it is not working. </p>
  
  <p>Producer code:</p>
  
  <pre><code>public static final String schema = "{"
      +"\"fields\": ["
      +   " { \"name\": \"str1\", \"type\": \"string\" },"
      +   " { \"name\": \"str2\", \"type\": \"string\" },"
      +   " { \"name\": \"int1\", \"type\": \"int\" }"
      +"],"
      +"\"name\": \"myrecord\","
      +"\"type\": \"record\""
      +"}"; 
  
      public static void startAvroProducer() throws InterruptedException, IOException{
          Properties props = new Properties();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
          props.put(ProducerConfig.CLIENT_ID_CONFIG, "Kafka Avro Producer");
  
          Schema.Parser parser = new Schema.Parser();
          Schema schema = parser.parse(AvroProducer.schema);
  
          AvroKey&lt;GenericRecord&gt; k = new AvroKey&lt;GenericRecord&gt;();
  
          GenericRecord datum = new GenericData.Record(schema);
          datum.put("str1","phani");
          datum.put("str2", "kumar");
          datum.put("int1", 1);
          k.datum(datum);
          GenericDatumWriter&lt;GenericRecord&gt; writer = new GenericDatumWriter&lt;GenericRecord&gt;(schema);
          ByteArrayOutputStream os = new ByteArrayOutputStream();
          Encoder e = EncoderFactory.get().binaryEncoder(os, null);
          writer.write(k.datum(), e);
          e.flush();
          byte[] bytedata = os.toByteArray();
          KafkaProducer&lt;String,byte[]&gt; producer = new KafkaProducer&lt;String,byte[]&gt;(props);
          ProducerRecord&lt;String,byte[]&gt; producerRec = new ProducerRecord&lt;String, byte[]&gt;("jason", bytedata);
          producer.send(producerRec);
          producer.close();
      }
  </code></pre>
  
  <p>Consumer code:</p>
  
  <pre><code>private static SparkConf sc = null;
  private static JavaSparkContext jsc = null;
  private static JavaStreamingContext jssc = null;
  
  public static void startAvroConsumer() throws InterruptedException {
      sc = new SparkConf().setAppName("Spark Avro Streaming Consumer")
              .setMaster("local[*]");
  
      jsc = new JavaSparkContext(sc);
      jssc = new JavaStreamingContext(jsc, new Duration(200));
  
      Schema.Parser parser = new Schema.Parser();
      Schema schema = parser.parse(AvroProducer.schema);
  
      Set&lt;String&gt; topics = Collections.singleton("jason");
      Map&lt;String, String&gt; kafkaParams = new HashMap&lt;String, String&gt;();
      kafkaParams.put("metadata.broker.list", "localhost:9092");
      kafkaParams.put("key.deserializer",
              "org.apache.kafka.common.serialization.StringDeserializer");
      kafkaParams.put("value.deserializer",
              "org.apache.kafka.common.serialization.ByteArrayDeserializer");
  
      JavaPairInputDStream&lt;String, byte[]&gt; inputDstream = KafkaUtils
              .createDirectStream(jssc, String.class, byte[].class,
                      StringDecoder.class, DefaultDecoder.class, kafkaParams,
                      topics);
  
      GenericDatumReader&lt;GenericRecord&gt; reader = new GenericDatumReader&lt;GenericRecord&gt;(
              schema);
  
      inputDstream.map(message -&gt; {
          ByteArrayInputStream bis = new ByteArrayInputStream(message._2);
          Decoder decoder = DecoderFactory.get().binaryDecoder(bis, null);
          GenericRecord record = reader.read(null, decoder);
          String str1 = getValue(record, "str1", String.class);
          String str2 = getValue(record, "str2", String.class);
          int int1 = getValue(record, "int1", Integer.class);
  
          return str1 + " " + str2 + " " + int1;
      }).print();;
  
      jssc.start();
      jssc.awaitTermination();
  }
  
  @SuppressWarnings("unchecked")
  public static &lt;T&gt; T getValue(GenericRecord genericRecord, String name,
          Class&lt;T&gt; clazz) {
      Object obj = genericRecord.get(name);
      if (obj == null)
          return null;
      if (obj.getClass() == Utf8.class) {
          return (T) obj.toString();
      }
      if (obj.getClass() == Integer.class) {
          return (T) obj;
      }
      return null;
  }
  </code></pre>
  
  <p>Exception:</p>
  
  <pre><code>Caused by: java.io.NotSerializableException: org.apache.avro.generic.GenericDatumReader
  Serialization stack:
      - object not serializable (class: org.apache.avro.generic.GenericDatumReader, value: org.apache.avro.generic.GenericDatumReader@7da8db47)
      - element of array (index: 0)
      - array (class [Ljava.lang.Object;, size 1)
      - field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)
      - object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=class com.applications.streaming.consumers.AvroConsumer, functionalInterfaceMethod=org/apache/spark/api/java/function/Function.call:(Ljava/lang/Object;)Ljava/lang/Object;, implementation=invokeStatic com/applications/streaming/consumers/AvroConsumer.lambda$0:(Lorg/apache/avro/generic/GenericDatumReader;Lscala/Tuple2;)Ljava/lang/String;, instantiatedMethodType=(Lscala/Tuple2;)Ljava/lang/String;, numCaptured=1])
      - writeReplace data (class: java.lang.invoke.SerializedLambda)
      - object (class com.applications.streaming.consumers.AvroConsumer$$Lambda$13/1805404637, com.applications.streaming.consumers.AvroConsumer$$Lambda$13/1805404637@aa31e58)
      - field (class: org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, name: fun$1, type: interface org.apache.spark.api.java.function.Function)
      - object (class org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, &lt;function1&gt;)
      at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
      at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
      at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:295)
      ... 15 more
  </code></pre>
  
  <p>While reading through various blogs what I understood is Avro objects does not implement serializable interface. But, as per the below jira</p>
  
  <p><a href="https://issues.apache.org/jira/browse/AVRO-1502" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/AVRO-1502</a></p>
  
  <p>The issue is resolved.I am still having this issue.</p>
  
  <p>Is there a possible fix to this issue. </p>