class BulkIngestMapFileLoader implements Runnable {

    public void cleanUpJobDirectory(Path mapFilesDir) throws IOException {
        Path jobDirectory = mapFilesDir.getParent();
        
        FileSystem destFs = getFileSystem(destHdfs);
        
        FileStatus[] failedDirs = destFs.globStatus(new Path(mapFilesDir, "failures/*/*"));
        boolean jobSucceeded = failedDirs == null || failedDirs.length == 0;
        if (jobSucceeded) {
            markSourceFilesLoaded(jobDirectory);
            // delete the successfully loaded map files directory and its parent directory
            destFs.delete(jobDirectory, true);
        } else {
            log.error("There were failures bringing map files online.  See: failed." + mapFilesDir.getName() + "failures/* for details");
            
            // rename the map files directory
            boolean success = destFs.rename(mapFilesDir, new Path(mapFilesDir.getParent(), "failed." + mapFilesDir.getName()));
            if (!success)
                log.error("Unable to rename map files directory " + destFs.getUri() + " " + mapFilesDir + " to failed." + mapFilesDir.getName());
            
            // create the job.failed file (renamed from job.loading if possible)
            success = destFs.rename(new Path(jobDirectory, LOADING_FILE_MARKER), new Path(jobDirectory, FAILED_FILE_MARKER));
            if (!success) {
                success = destFs.createNewFile(new Path(jobDirectory, FAILED_FILE_MARKER));
                if (!success)
                    log.error("Unable to create " + FAILED_FILE_MARKER + " file in " + jobDirectory);
            }
        }
        
        if (cleanUpScript != null) {
            Process proc = Runtime.getRuntime().exec(new String[] {cleanUpScript, jobDirectory.toString(), "" + jobSucceeded});
            int code;
            try {
                code = proc.waitFor();
            } catch (InterruptedException ex) {
                log.error("Cleanup script interrupted. ");
                code = -1;
            }
            if (code != 0) {
                log.error("Error occurred running external cleanup script: " + cleanUpScript);
            }
        }
        
    }

    public  BulkIngestMapFileLoader(String workDir, String jobDirPattern, String instanceName, String zooKeepers, Credentials credentials, URI seqFileHdfs,
                    URI srcHdfs, URI destHdfs, String jobtracker, Map<String,Integer> tablePriorities, Configuration conf);
    public  BulkIngestMapFileLoader(String workDir, String jobDirPattern, String instanceName, String zooKeepers, Credentials credentials, URI seqFileHdfs,
                    URI srcHdfs, URI destHdfs, String jobtracker, Map<String,Integer> tablePriorities, Configuration conf, int shutdownPort);
    public  BulkIngestMapFileLoader(String workDir, String jobDirPattern, String instanceName, String zooKeepers, Credentials credentials, URI seqFileHdfs,
                    URI srcHdfs, URI destHdfs, String jobtracker, Map<String,Integer> tablePriorities, Configuration conf, int shutdownPort,
                    int numHdfsThreads, List<Observer> jobObservers);

}