@SuppressWarnings("unchecked")
  public <IN, OUT> AvroDatumConverter<IN, OUT> create(Class<IN> inputClass) {
    boolean isMapOnly = ((JobConf) getConf()).getNumReduceTasks() == 0;
    if (AvroKey.class.isAssignableFrom(inputClass)) {
      Schema schema;
      if (isMapOnly) {
        schema = AvroJob.getMapOutputKeySchema(getConf());
        if (null == schema) {
          schema = AvroJob.getOutputKeySchema(getConf());
        }
      } else {
        schema = AvroJob.getOutputKeySchema(getConf());
      }
      if (null == schema) {
        throw new IllegalStateException("Writer schema for output key was not set. Use AvroJob.setOutputKeySchema().");
      }
      return (AvroDatumConverter<IN, OUT>) new AvroWrapperConverter(schema);
    }
    if (AvroValue.class.isAssignableFrom(inputClass)) {
      Schema schema;
      if (isMapOnly) {
        schema = AvroJob.getMapOutputValueSchema(getConf());
        if (null == schema) {
          schema = AvroJob.getOutputValueSchema(getConf());
        }
      } else {
        schema = AvroJob.getOutputValueSchema(getConf());
      }
      if (null == schema) {
        throw new IllegalStateException(
            "Writer schema for output value was not set. Use AvroJob.setOutputValueSchema().");
      }
      return (AvroDatumConverter<IN, OUT>) new AvroWrapperConverter(schema);
    }
    if (BooleanWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new BooleanWritableConverter();
    }
    if (BytesWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new BytesWritableConverter();
    }
    if (ByteWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new ByteWritableConverter();
    }
    if (DoubleWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new DoubleWritableConverter();
    }
    if (FloatWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new FloatWritableConverter();
    }
    if (IntWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new IntWritableConverter();
    }
    if (LongWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new LongWritableConverter();
    }
    if (NullWritable.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new NullWritableConverter();
    }
    if (Text.class.isAssignableFrom(inputClass)) {
      return (AvroDatumConverter<IN, OUT>) new TextConverter();
    }

    throw new UnsupportedOperationException("Unsupported input type: " + inputClass.getName());
  }
@Test
  void convertLongWritable() {
    AvroDatumConverter<LongWritable, Long> converter = mFactory.create(LongWritable.class);
    assertEquals(123L, converter.convert(new LongWritable(123L)).longValue());
  }