@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);
  }
@Test
  void getSerializerForKey() throws IOException {
    // Set the writer schema in the job configuration.
    Schema writerSchema = Schema.create(Schema.Type.STRING);
    Job job = Job.getInstance();
    AvroJob.setMapOutputKeySchema(job, writerSchema);

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

    // Check that the writer schema is set correctly on the serializer.
    assertEquals(writerSchema, avroSerializer.getWriterSchema());
  }