static CodecFactory getCodecFactory(JobConf job) {
    CodecFactory factory = null;

    if (FileOutputFormat.getCompressOutput(job)) {
      int deflateLevel = job.getInt(DEFLATE_LEVEL_KEY, DEFAULT_DEFLATE_LEVEL);
      int xzLevel = job.getInt(XZ_LEVEL_KEY, DEFAULT_XZ_LEVEL);
      int zstdLevel = job.getInt(ZSTD_LEVEL_KEY, DEFAULT_ZSTANDARD_LEVEL);
      boolean zstdBufferPool = job.getBoolean(ZSTD_BUFFERPOOL_KEY, DEFAULT_ZSTANDARD_BUFFERPOOL);
      String codecName = job.get(AvroJob.OUTPUT_CODEC);

      if (codecName == null) {
        String codecClassName = job.get("mapred.output.compression.codec", null);
        String avroCodecName = HadoopCodecFactory.getAvroCodecName(codecClassName);
        if (codecClassName != null && avroCodecName != null) {
          factory = HadoopCodecFactory.fromHadoopString(codecClassName);
          job.set(AvroJob.OUTPUT_CODEC, avroCodecName);
          return factory;
        } else {
          return CodecFactory.deflateCodec(deflateLevel);
        }
      } else {
        if (codecName.equals(DEFLATE_CODEC)) {
          factory = CodecFactory.deflateCodec(deflateLevel);
        } else if (codecName.equals(XZ_CODEC)) {
          factory = CodecFactory.xzCodec(xzLevel);
        } else if (codecName.equals(ZSTANDARD_CODEC)) {
          factory = CodecFactory.zstandardCodec(zstdLevel, false, zstdBufferPool);
        } else {
          factory = CodecFactory.fromString(codecName);
        }
      }
    }

    return factory;
  }
@Test
  void deflateCodecUsingHadoopClass() {
    CodecFactory avroDeflateCodec = CodecFactory.fromString("deflate");

    JobConf job = new JobConf();
    job.set("mapred.output.compress", "true");
    job.set("mapred.output.compression.codec", "org.apache.hadoop.io.compress.DeflateCodec");
    CodecFactory factory = AvroOutputFormat.getCodecFactory(job);
    assertNotNull(factory);
    assertEquals(factory.getClass(), avroDeflateCodec.getClass());
  }