### Question:
AvroSerialization extends Configured implements Serialization<AvroWrapper<T>> { @Override public Serializer<AvroWrapper<T>> getSerializer(Class<AvroWrapper<T>> c) { Configuration conf = getConf(); Schema schema; if (AvroKey.class.isAssignableFrom(c)) { schema = getKeyWriterSchema(conf); } else if (AvroValue.class.isAssignableFrom(c)) { schema = getValueWriterSchema(conf); } else { throw new IllegalStateException("Only AvroKey and AvroValue are supported."); } GenericData dataModel = createDataModel(conf); DatumWriter<T> datumWriter = dataModel.createDatumWriter(schema); return new AvroSerializer<>(schema, datumWriter); }  @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 testGetSerializerForKey() throws IOException { Schema writerSchema = Schema.create(Schema.Type.STRING); Job job = Job.getInstance(); AvroJob.setMapOutputKeySchema(job, writerSchema); AvroSerialization serialization = ReflectionUtils.newInstance(AvroSerialization.class, job.getConfiguration()); @SuppressWarnings("unchecked") Serializer<AvroWrapper> serializer = serialization.getSerializer(AvroKey.class); assertTrue(serializer instanceof AvroSerializer); AvroSerializer avroSerializer = (AvroSerializer) serializer; assertEquals(writerSchema, avroSerializer.getWriterSchema()); }
@Test public void testGetSerializerForValue() throws IOException { Schema writerSchema = Schema.create(Schema.Type.STRING); Job job = Job.getInstance(); AvroJob.setMapOutputValueSchema(job, writerSchema); AvroSerialization serialization = ReflectionUtils.newInstance(AvroSerialization.class, job.getConfiguration()); @SuppressWarnings("unchecked") Serializer<AvroWrapper> serializer = serialization.getSerializer(AvroValue.class); assertTrue(serializer instanceof AvroSerializer); AvroSerializer avroSerializer = (AvroSerializer) serializer; assertEquals(writerSchema, avroSerializer.getWriterSchema()); }