@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 getDeserializerForValue() throws IOException {
    // Set the reader schema in the job configuration.
    Schema readerSchema = Schema.create(Schema.Type.STRING);
    Job job = Job.getInstance();
    AvroJob.setMapOutputValueSchema(job, readerSchema);

    // Get a deserializer from the configuration.
    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;

    // Check that the reader schema is set correctly on the deserializer.
    assertEquals(readerSchema, avroDeserializer.getReaderSchema());
  }