public static PathOutputCommitterFactory getCommitterFactory(
      Path outputPath,
      Configuration conf) {
    // determine which key to look up the overall one or a schema-specific
    // key
    LOG.debug("Looking for committer factory for path {}", outputPath);
    String key = COMMITTER_FACTORY_CLASS;
    if (StringUtils.isEmpty(conf.getTrimmed(key)) && outputPath != null) {
      // there is no explicit factory and there's an output path
      // Get the scheme of the destination
      String scheme = outputPath.toUri().getScheme();

      // and see if it has a key
      String schemeKey = String.format(COMMITTER_FACTORY_SCHEME_PATTERN,
          scheme);
      if (StringUtils.isNotEmpty(conf.getTrimmed(schemeKey))) {
        // it does, so use that key in the classname lookup
        LOG.info("Using schema-specific factory for {}", outputPath);
        key = schemeKey;
      } else {
        LOG.debug("No scheme-specific factory defined in {}", schemeKey);
      }
    }

    // create the factory. Before using Configuration.getClass, check
    // for an empty configuration value, as that raises ClassNotFoundException.
    Class<? extends PathOutputCommitterFactory> factory;
    String trimmedValue = conf.getTrimmed(key, "");
    if (StringUtils.isEmpty(trimmedValue)) {
      // empty/null value, use default
      LOG.info("No output committer factory defined,"
          + " defaulting to FileOutputCommitterFactory");
      factory = FileOutputCommitterFactory.class;
    } else {
      // key is set, get the class
      factory = conf.getClass(key,
          FileOutputCommitterFactory.class,
          PathOutputCommitterFactory.class);
      LOG.info("Using OutputCommitter factory class {} from key {}",
          factory, key);
    }
    return ReflectionUtils.newInstance(factory, conf);
  }
@Test
  public void testCommitterFactoryUnknown() throws Throwable {
    Configuration conf = new Configuration();
    // set the factory to an unknown class
    conf.set(COMMITTER_FACTORY_CLASS, "unknown");
    intercept(RuntimeException.class,
        () -> getCommitterFactory(HDFS_PATH, conf));
  }