title: Error in serialization when consuming JSON Objects from Kafka
body: <p>I am trying to produce Json Objects to Kafka and consume them manually, I am using the JSONPOJO Serdes in org.apache.kafka.streams.examples.pageview.</p>
  
  <p>My producer code is : </p>
  
  <pre><code>package JsonProducer;
  
  imports ...
  
  public class jsnPdc {
  
      public static void main(String[] args) throws IOException {
  
          byte[] arr= "XXXX       THIS IS TEST DATA \n XYZ".getBytes();    
          JSONObject jsn = new JSONObject();
          jsn.put("Header_Title", (Arrays.copyOfRange(arr, 0, 4)));
          jsn.put("Data_Part", (Arrays.copyOfRange(arr, 4, arr.length)));
  
  
          Properties props = new Properties();
          props.put("bootstrap.servers", "xxxxxxxxxxxxxxxxxxxxx:xxxx");
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer", "org.apache.kafka.streams.examples.pageview.JsonPOJOSerializer");
  
          KafkaProducer&lt;String, JSONObject&gt; pdc = new KafkaProducer&lt;&gt;(props);
          pdc.send(new ProducerRecord&lt;String,JSONObject&gt;("testoutput", jsn));
  
          System.in.read();
  
  
      }
  
  }
  </code></pre>
  
  <p>and the code for consumer is :</p>
  
  <pre><code>package testConsumer;
  
  imports ...
  
  public class consumer_0 {
      static public void main(String[] argv) throws ParseException {
  
          //Configuration
          Properties props = new Properties();
          props.put("bootstrap.servers", "xxxxxxxxxxxxxxxxxxx:xxxx");
          props.put("group.id", "test");
          props.put("enable.auto.commit", "false");
          props.put("auto.commit.interval.ms", "1000");
          props.put("session.timeout.ms", "30000");
          props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          props.put("value.deserializer", "org.apache.kafka.streams.examples.pageview.JsonPOJODeserializer");
  
  
          //Create Consumer Object
          KafkaConsumer&lt;String, JSONObject&gt; consumer = new KafkaConsumer&lt;String, JSONObject&gt;(props);
          consumer.subscribe(Arrays.asList("testoutput"));
  
  
          //Keep Polling Records
          System.out.println("Polling new record...\n");
          while (true) {
              ConsumerRecords&lt;String, JSONObject&gt; records = consumer.poll(100);
  
              //Print Each Record
              for (ConsumerRecord&lt;String, JSONObject&gt; record : records){
                  JSONObject json = record.value();
  
                  //Some print code, print(json) ...
  
              }
          }
      }
  }
  </code></pre>
  
  <p>And I get this problem: </p>
  
  <pre><code>Exception in thread "main" org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition testoutput-0 at offset 20491
  Caused by: org.apache.kafka.common.errors.SerializationException: java.lang.IllegalArgumentException: Unrecognized Type: [null]
  Caused by: java.lang.IllegalArgumentException: Unrecognized Type: [null]
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1170)
      at com.fasterxml.jackson.databind.type.TypeFactory.constructType(TypeFactory.java:618)
      at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2929)
      at org.apache.kafka.streams.examples.pageview.JsonPOJODeserializer.deserialize(JsonPOJODeserializer.java:49)
      at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:882)
      at org.apache.kafka.clients.consumer.internals.Fetcher.parseCompletedFetch(Fetcher.java:788)
      at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:480)
      at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1061)
      at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995)
      at testConsumer.consumer_0.main(consumer_0.java:43) 
  </code></pre>
  
  <p>I need the value field type of json to be in byte array. Any idea why this is happening? </p>