title: How to discard default value fields of AVRO data at the Kafka AVRO consumer end?
body: <p>I have defined schema in the schema registry with 10 fields. 
  Using  <strong>confluent-3.3.0</strong>, <strong>Kafka 0.10</strong></p>
  
  <pre><code>{"schema": "{\"type\":\"record\",\"name\":\"User2\",\"fields\":[{\"name\":\"userName\",\"type\":\"string\"},{\"name\":\"uID\",\"type\":\"string\"},{\"name\":\"company\",\"type\":\"string\",\"default\":\"ABC\"},{\"name\":\"age\",\"type\":\"int\",\"default\":0},{\"name\":\"number\",\"type\":\"int\",\"default\":0},{\"name\":\"firstCompany\",\"type\":\"string\",\"default\":\"\"},{\"name\":\"lastCompany\",\"type\":\"string\",\"default\":\"\"},{\"name\":\"mobileNumber\",\"type\":\"int\",\"default\":0},{\"name\":\"email\",\"type\":\"string\",\"default\":\"\"},{\"name\":\"birthYear\",\"type\":\"int\",\"default\":1980}]}"}
  </code></pre>
  
  <p>Fields without default value: <strong>userName,uID</strong>
  Fields with default value: <strong>company,age,number,firstCompany,lastCompany,mobileNumber,email,birthYear</strong>.</p>
  
  <p>Kafka Producer code is given below. I am passing only 2 fields <strong>userName,uID</strong> to the Kafka producer i.e. AvroProducer</p>
  
  <pre><code> public class AvroProducer {
   public static void main(String[] args) throws ExecutionException, InterruptedException {
  
          Properties props = new Properties();
          props.put("bootstrap.servers", "localhost:9092");
          props.put("ZOOKEEPER_HOST", "localhost");
          //props.put("acks", "all");
          props.put("retries", 0);
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
          props.put("schema.registry.url", "http://localhost:8081");
          String topic = "confluent-new";
  
          Schema.Parser parser = new Schema.Parser();
          // I will get below schema string from SCHEMA REGISTRY
          Schema schema = parser.parse("{\"type\":\"record\",\"name\":\"User2\",\"fields\":[{\"name\":\"userName\",\"type\":\"string\"},{\"name\":\"uID\",\"type\":\"string\"},{\"name\":\"company\",\"type\":\"string\",\"default\":\"ABC\"},{\"name\":\"age\",\"type\":\"int\",\"default\":0},{\"name\":\"number\",\"type\":\"int\",\"default\":0},{\"name\":\"firstCompany\",\"type\":\"string\",\"default\":\"\"},{\"name\":\"lastCompany\",\"type\":\"string\",\"default\":\"\"},{\"name\":\"mobileNumber\",\"type\":\"int\",\"default\":0},{\"name\":\"email\",\"type\":\"string\",\"default\":\"\"},{\"name\":\"birthYear\",\"type\":\"int\",\"default\":1980}]}");
  
          Producer&lt;String, GenericRecord&gt; producer = new KafkaProducer&lt;String, GenericRecord&gt;(props);
          GenericRecord record = new GenericData.Record(schema);
          record.put("uID", "06080000");
          record.put("userName", "User data10");
  
          ProducerRecord&lt;String, GenericRecord&gt; recordData = new ProducerRecord&lt;String, GenericRecord&gt;(topic, "ip", record);
          producer.send(recordData);
  
          System.out.println("Message Sent");
      }
  </code></pre>
  
  <p>And the Consumer code is</p>
  
  <pre><code>public class AvroConsumer {
  public static void main(String[] args) throws ExecutionException, InterruptedException {
  
  Properties props = new Properties();
  props.put("bootstrap.servers", "localhost:9092");
  props.put("ZOOKEEPER_HOST", "localhost");
  props.put("acks", "all");
  props.put("retries", 0);
  props.put("group.id", "consumer1");
  props.put("auto.offset.reset", "latest");
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
  props.put("schema.registry.url", "http://localhost:8081");
  String topic = "confluent-new";
  
  KafkaConsumer&lt;String, GenericRecord&gt; consumer = new KafkaConsumer&lt;String, GenericRecord&gt;(props);
  consumer.subscribe(Arrays.asList(topic));
  while(true){
      ConsumerRecords&lt;String, GenericRecord&gt; recs = consumer.poll(10000);
      for (ConsumerRecord&lt;String, GenericRecord&gt; rec : recs) {
          System.out.printf("{AvroConsumer}: Recieved [key= %s, value= %s]\n", rec.key(), rec.value());
      }
  }
  }
  </code></pre>
  
  <p>While running the above code I get below exception at Producer end.</p>
  
  <pre><code> log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
  Exception in thread "main" org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
  Caused by: java.lang.NullPointerException: null of string in field location of http
      at org.apache.avro.generic.GenericDatumWriter.npe(GenericDatumWriter.java:145)
      at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:139)
      at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
      at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:62)
      at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:92)
      at kafka.CTKafkaAvroSerializer.serialize(CTKafkaAvroSerializer.java:12)
      at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:453)
      at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:430)
      at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:353)
      at kafka.AvroProducerWithJSON.main(AvroProducerWithJSON.java:64)
  Caused by: java.lang.NullPointerException
      at org.apache.avro.io.Encoder.writeString(Encoder.java:121)
      at org.apache.avro.generic.GenericDatumWriter.writeString(GenericDatumWriter.java:267)
  </code></pre>
  
  <p>To overcome above exception I have populated the default fields with values from schema itself viz. <strong>company,age,number,firstCompany,lastCompany,mobileNumber,email,birthYear</strong> at Producer end. Now I am getting output at Consumer end like below.</p>
  
  <pre><code>{AvroConsumer}: Recieved [key= ip, value= {"userName": "User data10", "uID": "06080000", "company": "ABC", "age": 0, "number": 0,"firstCompany": "","lastCompany": "","mobileNumber": 0,"email": "","birthYear": 1980}]
  </code></pre>
  
  <p>So I am getting default values of all other(unintended) fields. I want to avoid this. I want my output should only contains those fields which I was intended to pass (in this case <strong>userName, uID</strong>) and discard all default value fields.</p>
  
  <p>Any pointer around this will be greatly appreciated.</p>