### Question:
AvroSerialization extends Configured implements Serialization<AvroWrapper<T>> { @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."); } }  @Override boolean accept(Class<?> c); @Override Deserializer<AvroWrapper<T>> getDeserializer(Class<AvroWrapper<T>> c); @Override Serializer<AvroWrapper<T>> getSerializer(Class<AvroWrapper<T>> c); static void addToConfiguration(Configuration conf); static void setKeyWriterSchema(Configuration conf, Schema schema); static void setKeyReaderSchema(Configuration conf, Schema schema); static void setValueWriterSchema(Configuration conf, Schema schema); static void setValueReaderSchema(Configuration conf, Schema schema); static void setDataModelClass(Configuration conf, Class<? extends GenericData> modelClass); static Schema getKeyWriterSchema(Configuration conf); static Schema getKeyReaderSchema(Configuration conf); static Schema getValueWriterSchema(Configuration conf); static Schema getValueReaderSchema(Configuration conf); static Class<? extends GenericData> getDataModelClass(Configuration conf); static GenericData createDataModel(Configuration conf);  }### Answer:
@Test public void testGetDeserializerForKey() throws IOException { Schema readerSchema = Schema.create(Schema.Type.STRING); Job job = Job.getInstance(); AvroJob.setMapOutputKeySchema(job, readerSchema); AvroSerialization serialization = ReflectionUtils.newInstance(AvroSerialization.class, job.getConfiguration()); @SuppressWarnings("unchecked") Deserializer<AvroWrapper> deserializer = serialization.getDeserializer(AvroKey.class); assertTrue(deserializer instanceof AvroKeyDeserializer); AvroKeyDeserializer avroDeserializer = (AvroKeyDeserializer) deserializer; assertEquals(readerSchema, avroDeserializer.getReaderSchema()); }
@Test public void testGetDeserializerForValue() throws IOException { Schema readerSchema = Schema.create(Schema.Type.STRING); Job job = Job.getInstance(); AvroJob.setMapOutputValueSchema(job, readerSchema); AvroSerialization serialization = ReflectionUtils.newInstance(AvroSerialization.class, job.getConfiguration()); @SuppressWarnings("unchecked") Deserializer<AvroWrapper> deserializer = serialization.getDeserializer(AvroValue.class); assertTrue(deserializer instanceof AvroValueDeserializer); AvroValueDeserializer avroDeserializer = (AvroValueDeserializer) deserializer; assertEquals(readerSchema, avroDeserializer.getReaderSchema()); }
@Test public void testClassPath() 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(); Assert.assertEquals(loader, data.getClassLoader()); }