public KafkaAvroSchemaDeserializer() {
  }
@Test
  public void testKafkaAvroSchemaDeserializer() {
    byte[] bytesOrigRecord;
    IndexedRecord avroRecord = createUserRecord();
    config.put(AvroKafkaSource.KAFKA_AVRO_VALUE_DESERIALIZER_SCHEMA, origSchema.toString());

    KafkaAvroSchemaDeserializer avroDeserializer = new KafkaAvroSchemaDeserializer(schemaRegistry, new HashMap(config));
    avroDeserializer.configure(new HashMap(config), false);
    bytesOrigRecord = avroSerializer.serialize(topic, avroRecord);
    // record is serialized in orig schema and deserialized using same schema.
    assertEquals(avroRecord, avroDeserializer.deserialize(topic, false, bytesOrigRecord, origSchema));

    IndexedRecord avroRecordWithAllField = createExtendUserRecord();
    byte[] bytesExtendedRecord = avroSerializer.serialize(topic, avroRecordWithAllField);

    SchemaTestProvider.schemaToReturn.set(evolSchema);
    config.put(AvroKafkaSource.KAFKA_AVRO_VALUE_DESERIALIZER_SCHEMA, evolSchema.toString());
    avroDeserializer = new KafkaAvroSchemaDeserializer(schemaRegistry, new HashMap(config));
    avroDeserializer.configure(new HashMap(config), false);
    // record is serialized w/ evolved schema, and deserialized w/ evolved schema
    IndexedRecord avroRecordWithAllFieldActual = (IndexedRecord) avroDeserializer.deserialize(topic, false, bytesExtendedRecord, evolSchema);
    assertEquals(avroRecordWithAllField, avroRecordWithAllFieldActual);
    assertEquals(avroRecordWithAllFieldActual.getSchema(), evolSchema);

    // read old record w/ evolved schema.
    IndexedRecord actualRec = (IndexedRecord) avroDeserializer.deserialize(topic, false, bytesOrigRecord, origSchema);
    // record won't be equal to original record as we read w/ evolved schema. "age" will be added w/ default value of null
    assertNotEquals(avroRecord, actualRec);
    GenericRecord genericRecord = (GenericRecord) actualRec;
    GenericRecord origGenRec = (GenericRecord) avroRecord;
    assertEquals(genericRecord.get("name").toString(), origGenRec.get("name").toString());
    assertEquals(actualRec.getSchema(), evolSchema);
    assertNull(genericRecord.get("age"));
  }