@Test
public void testFragmenterErrorsWhenPathDoesNotExist() throws Exception {
    expectedException.expect(InvalidInputException.class);
    expectedException.expectMessage("Input path does not exist:");

    String path = this.getClass().getClassLoader().getResource("csv/").getPath();

    RequestContext context = new RequestContext();
    context.setConfig("default");
    context.setUser("test-user");
    context.setProfileScheme("localfile");
    context.setDataSource(path + "non-existent");

    Fragmenter fragmenter = new HdfsFileFragmenter();
    fragmenter.initialize(context);
    fragmenter.getFragments();
}
 

/** Sanity check for source */
protected static void checkSource(Configuration conf, List<Path> srcs
    ) throws InvalidInputException {
  List<IOException> ioes = new ArrayList<IOException>();
  for(Path p : srcs) {
    try {
      if (!p.getFileSystem(conf).exists(p)) {
        ioes.add(new FileNotFoundException("Source "+p+" does not exist."));
      }
    }
    catch(IOException e) {ioes.add(e);}
  }
  if (!ioes.isEmpty()) {
    throw new InvalidInputException(ioes);
  }
}
 

/** Sanity check for source */
protected static void checkSource(Configuration conf, List<Path> srcs
    ) throws InvalidInputException {
  List<IOException> ioes = new ArrayList<IOException>();
  for(Path p : srcs) {
    try {
      if (!p.getFileSystem(conf).exists(p)) {
        ioes.add(new FileNotFoundException("Source "+p+" does not exist."));
      }
    }
    catch(IOException e) {ioes.add(e);}
  }
  if (!ioes.isEmpty()) {
    throw new InvalidInputException(ioes);
  }
}
 

/** Sanity check for source */
protected static void checkSource(Configuration conf, List<Path> srcs
    ) throws InvalidInputException {
  List<IOException> ioes = new ArrayList<IOException>();
  for(Path p : srcs) {
    try {
      if (!p.getFileSystem(conf).exists(p)) {
        ioes.add(new FileNotFoundException("Source "+p+" does not exist."));
      }
    }
    catch(IOException e) {ioes.add(e);}
  }
  if (!ioes.isEmpty()) {
    throw new InvalidInputException(ioes);
  }
}
 

/** Sanity check for srcPath */
private static void checkSrcPath(Configuration conf, List<Path> srcPaths
    ) throws IOException {
  List<IOException> rslt = new ArrayList<IOException>();
  List<Path> unglobbed = new LinkedList<Path>();
  for (Path p : srcPaths) {
    FileSystem fs = p.getFileSystem(conf);
    FileStatus[] inputs = fs.globStatus(p);

    if(inputs.length > 0) {
      for (FileStatus onePath: inputs) {
        unglobbed.add(onePath.getPath());
      }
    } else {
      rslt.add(new IOException("Input source " + p + " does not exist."));
    }
  }
  if (!rslt.isEmpty()) {
    throw new InvalidInputException(rslt);
  }
  srcPaths.clear();
  srcPaths.addAll(unglobbed);
}
 

/** Sanity check for source */
protected static void checkSource(Configuration conf, List<Path> srcs
    ) throws InvalidInputException {
  List<IOException> ioes = new ArrayList<IOException>();
  for(Path p : srcs) {
    try {
      if (!p.getFileSystem(conf).exists(p)) {
        ioes.add(new FileNotFoundException("Source "+p+" does not exist."));
      }
    }
    catch(IOException e) {ioes.add(e);}
  }
  if (!ioes.isEmpty()) {
    throw new InvalidInputException(ioes);
  }
}
 

/**
 * Gets the fragments for a data source URI that can appear as a file name,
 * a directory name or a wildcard. Returns the data fragments in JSON
 * format.
 */
@Override
public List<Fragment> getFragments() throws Exception {
    String fileName = hcfsType.getDataUri(jobConf, context);
    Path path = new Path(fileName);

    PxfInputFormat pxfInputFormat = new PxfInputFormat();
    PxfInputFormat.setInputPaths(jobConf, path);

    FileStatus[] fileStatusArray;

    try {
        fileStatusArray = pxfInputFormat.listStatus(jobConf);
    } catch (InvalidInputException e) {
        if (StringUtils.equalsIgnoreCase("true", context.getOption(IGNORE_MISSING_PATH_OPTION))) {
            LOG.debug("Ignoring InvalidInputException", e);
            return fragments;
        }
        throw e;
    }

    fragments = Arrays.stream(fileStatusArray)
            .map(fileStatus -> new Fragment(fileStatus.getPath().toUri().toString()))
            .collect(Collectors.toList());
    LOG.debug("Total number of fragments = {}", fragments.size());

    return fragments;
}
 

/**
 * Gets the fragments for a data source URI that can appear as a file name,
 * a directory name or a wildcard. Returns the data fragments in JSON
 * format.
 */
@Override
public List<Fragment> getFragments() throws Exception {
    Path path = new Path(hcfsType.getDataUri(jobConf, context));
    List<InputSplit> splits;
    try {
        splits = getSplits(path);
    } catch (InvalidInputException e) {
        if (StringUtils.equalsIgnoreCase("true", context.getOption(IGNORE_MISSING_PATH_OPTION))) {
            LOG.debug("Ignoring InvalidInputException", e);
            return fragments;
        }
        throw e;
    }

    LOG.debug("Total number of fragments = {}", splits.size());
    for (InputSplit split : splits) {
        FileSplit fsp = (FileSplit) split;
        String filepath = fsp.getPath().toString();
        String[] hosts = fsp.getLocations();

        /*
         * metadata information includes: file split's start, length and
         * hosts (locations).
         */
        byte[] fragmentMetadata = HdfsUtilities.prepareFragmentMetadata(fsp);
        Fragment fragment = new Fragment(filepath, hosts, fragmentMetadata);
        fragments.add(fragment);
    }

    return fragments;
}
 

@Test
public void testInvalidInputPath() throws Exception {
    thrown.expect(InvalidInputException.class);
    thrown.expectMessage("Input Pattern file:/tmp/non-existent-path-on-disk/*.csv matches 0 files");

    RequestContext context = new RequestContext();
    context.setConfig("default");
    context.setUser("test-user");
    context.setProfileScheme("localfile");
    context.setDataSource("/tmp/non-existent-path-on-disk/*.csv");

    Fragmenter fragmenter = new HdfsDataFragmenter();
    fragmenter.initialize(context);
    fragmenter.getFragments();
}
 

@Test
public void testInvalidInputPath() throws Exception {
    expectedException.expect(InvalidInputException.class);
    expectedException.expectMessage("Input Pattern file:/tmp/non-existent-path-on-disk/*.csv matches 0 files");

    RequestContext context = new RequestContext();
    context.setConfig("default");
    context.setUser("test-user");
    context.setProfileScheme("localfile");
    context.setDataSource("/tmp/non-existent-path-on-disk/*.csv");

    Fragmenter fragmenter = new HdfsFileFragmenter();
    fragmenter.initialize(context);
    fragmenter.getFragments();
}
 

private static void check(Configuration conf, List<FileOperation> ops
    ) throws InvalidInputException {
  List<Path> srcs = new ArrayList<Path>();
  for(FileOperation op : ops) {
    srcs.add(op.src);
  }
  DistTool.checkSource(conf, srcs);
}
 

/** Sanity check for srcPath */
private static void checkSrcPath(JobConf jobConf, List<Path> srcPaths) 
throws IOException {
  List<IOException> rslt = new ArrayList<IOException>();
  List<Path> unglobbed = new LinkedList<Path>();
  
  Path[] ps = new Path[srcPaths.size()];
  ps = srcPaths.toArray(ps);
  TokenCache.obtainTokensForNamenodes(jobConf.getCredentials(), ps, jobConf);
  
  
  for (Path p : srcPaths) {
    FileSystem fs = p.getFileSystem(jobConf);
    FileStatus[] inputs = fs.globStatus(p);
    
    if(inputs != null && inputs.length > 0) {
      for (FileStatus onePath: inputs) {
        unglobbed.add(onePath.getPath());
      }
    } else {
      rslt.add(new IOException("Input source " + p + " does not exist."));
    }
  }
  if (!rslt.isEmpty()) {
    throw new InvalidInputException(rslt);
  }
  srcPaths.clear();
  srcPaths.addAll(unglobbed);
}
 

private static void check(Configuration conf, List<FileOperation> ops
    ) throws InvalidInputException {
  List<Path> srcs = new ArrayList<Path>();
  for(FileOperation op : ops) {
    srcs.add(op.src);
  }
  DistTool.checkSource(conf, srcs);
}
 

/** Sanity check for srcPath */
private static void checkSrcPath(JobConf jobConf, List<Path> srcPaths) 
throws IOException {
  List<IOException> rslt = new ArrayList<IOException>();
  List<Path> unglobbed = new LinkedList<Path>();
  
  Path[] ps = new Path[srcPaths.size()];
  ps = srcPaths.toArray(ps);
  TokenCache.obtainTokensForNamenodes(jobConf.getCredentials(), ps, jobConf);
  
  
  for (Path p : srcPaths) {
    FileSystem fs = p.getFileSystem(jobConf);
    FileStatus[] inputs = fs.globStatus(p);
    
    if(inputs != null && inputs.length > 0) {
      for (FileStatus onePath: inputs) {
        unglobbed.add(onePath.getPath());
      }
    } else {
      rslt.add(new IOException("Input source " + p + " does not exist."));
    }
  }
  if (!rslt.isEmpty()) {
    throw new InvalidInputException(rslt);
  }
  srcPaths.clear();
  srcPaths.addAll(unglobbed);
}
 

private static void check(Configuration conf, List<FileOperation> ops
    ) throws InvalidInputException {
  List<Path> srcs = new ArrayList<Path>();
  for(FileOperation op : ops) {
    srcs.add(op.src);
  }
  DistTool.checkSource(conf, srcs);
}
 

private static void check(Configuration conf, List<FileOperation> ops
    ) throws InvalidInputException {
  List<Path> srcs = new ArrayList<Path>();
  for(FileOperation op : ops) {
    srcs.add(op.src);
  }
  DistTool.checkSource(conf, srcs);
}
 

/** Sanity check for srcPath */
private static void checkSrcPath(Configuration conf, List<Path> srcPaths
    ) throws IOException {
  List<IOException> rslt = new ArrayList<IOException>();
  for (Path p : srcPaths) {
    FileSystem fs = p.getFileSystem(conf);
    if (!fs.exists(p)) {
      rslt.add(new IOException("Input source " + p + " does not exist."));
    }
  }
  if (!rslt.isEmpty()) {
    throw new InvalidInputException(rslt);
  }
}