@Override
  public Deserializer<AvroWrapper<T>> getDeserializer(Class<AvroWrapper<T>> c) {
    Configuration conf = getConf();
    GenericData dataModel = createDataModel(conf);
    if (AvroKey.class.isAssignableFrom(c)) {
      Schema writerSchema = getKeyWriterSchema(conf);
      Schema readerSchema = getKeyReaderSchema(conf);
      DatumReader<T> datumReader = (readerSchema != null) ? dataModel.createDatumReader(writerSchema, readerSchema)
          : dataModel.createDatumReader(writerSchema);
      return new AvroKeyDeserializer<>(writerSchema, readerSchema, datumReader);
    } else if (AvroValue.class.isAssignableFrom(c)) {
      Schema writerSchema = getValueWriterSchema(conf);
      Schema readerSchema = getValueReaderSchema(conf);
      DatumReader<T> datumReader = (readerSchema != null) ? dataModel.createDatumReader(writerSchema, readerSchema)
          : dataModel.createDatumReader(writerSchema);
      return new AvroValueDeserializer<>(writerSchema, readerSchema, datumReader);
    } else {
      throw new IllegalStateException("Only AvroKey and AvroValue are supported.");
    }
  }
@Test
  void classPath() throws Exception {
    Configuration conf = new Configuration();
    ClassLoader loader = conf.getClass().getClassLoader();
    AvroSerialization serialization = new AvroSerialization();
    serialization.setConf(conf);
    AvroDeserializer des = (AvroDeserializer) serialization.getDeserializer(AvroKey.class);
    ReflectData data = (ReflectData) ((ReflectDatumReader) des.mAvroDatumReader).getData();
    assertEquals(loader, data.getClassLoader());
  }