class BulkIngestMapFileLoader implements Runnable {

    public boolean takeOwnershipJobDirectory(Path jobDirectory) {
        boolean success = false;
        try {
            FileSystem fs = getFileSystem(srcHdfs);
            
            try {
                success = fs.rename(new Path(jobDirectory, COMPLETE_FILE_MARKER), new Path(jobDirectory, LOADING_FILE_MARKER));
                log.info("Renamed " + jobDirectory + '/' + COMPLETE_FILE_MARKER + " to " + LOADING_FILE_MARKER);
            } catch (IOException e2) {
                log.error("Exception while marking " + jobDirectory + " for loading: " + e2.getMessage(), e2);
            }
            
            // if not successful, see if we can provide a reason
            if (!success) {
                if (fs.exists(new Path(jobDirectory, LOADING_FILE_MARKER))) {
                    log.info("Another process already took ownership of " + jobDirectory + " for loading");
                } else {
                    log.error("Unable to take ownership of " + jobDirectory + " for loading");
                }
            } else {
                if (!fs.exists(new Path(jobDirectory, LOADING_FILE_MARKER))) {
                    // if the loading file marker does not exist, then we did not really succeed....hadoop strangeness?
                    log.error("Rename returned success but yet we did not take ownership of " + jobDirectory + " (" + LOADING_FILE_MARKER + " does not exist)");
                    success = false;
                } else if (fs.exists(new Path(jobDirectory, COMPLETE_FILE_MARKER))) {
                    // if the complete file still exists, then perhaps the IngestJob received a create failure and subsequently reattempted.
                    log.error("Rename returned success but yet we did not fully take ownership of " + jobDirectory + " (" + COMPLETE_FILE_MARKER + " moved to "
                                    + LOADING_FILE_MARKER + " but " + COMPLETE_FILE_MARKER + " still exists)");
                    success = false;
                    // move the job.loading out of the way. I don't want to delete any files just in case hadoop is getting confused
                    // and a delete might result in both files deleted and then we might think this is simply a failed distcp finally
                    // resulting in lost data.
                    int count = 0;
                    boolean done = false;
                    while (!done && fs.exists(new Path(jobDirectory, COMPLETE_FILE_MARKER)) && count < 10) {
                        count++;
                        if (fs.rename(new Path(jobDirectory, LOADING_FILE_MARKER), new Path(jobDirectory, ATTEMPT_FILE_MARKER + '.' + count))) {
                            log.error("Moved " + LOADING_FILE_MARKER + " to " + ATTEMPT_FILE_MARKER + '.' + count);
                            done = true;
                        }
                    }
                }
            }
        } catch (IOException e) {
            log.error("Exception while marking " + jobDirectory + " for loading: " + e.getMessage(), e);
        }
        return success;
    }

    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);

}