@Test(expected = RuntimeException.class) public void runJobWithErrorsAndFailOnErrors() throws Exception { props.setProperty(CamusJob.ETL_FAIL_ON_ERRORS, Boolean.TRUE.toString()); props.setProperty(EtlInputFormat.CAMUS_MESSAGE_DECODER_CLASS, FailDecoder.class.getName()); job = new CamusJob(props); job.run(); }
public void run() throws Exception { startTiming("pre-setup"); startTiming("total"); Job job = createJob(props); if (getLog4jConfigure(job)) { DOMConfigurator.configure("log4j.xml"); } FileSystem fs = FileSystem.get(job.getConfiguration()); log.info("Dir Destination set to: " + EtlMultiOutputFormat.getDestinationPath(job)); Path execBasePath = new Path(props.getProperty(ETL_EXECUTION_BASE_PATH)); Path execHistory = new Path(props.getProperty(ETL_EXECUTION_HISTORY_PATH)); if (!fs.exists(execBasePath)) { log.info("The execution base path does not exist. Creating the directory"); fs.mkdirs(execBasePath); } if (!fs.exists(execHistory)) { log.info("The history base path does not exist. Creating the directory."); fs.mkdirs(execHistory); } ContentSummary content = fs.getContentSummary(execBasePath); long limit = (long) (content.getQuota() * job.getConfiguration().getFloat(ETL_EXECUTION_HISTORY_MAX_OF_QUOTA, (float) .5)); limit = limit == 0 ? 50000 : limit; if (props.containsKey(ETL_BASEDIR_QUOTA_OVERIDE)) { limit = Long.valueOf(props.getProperty(ETL_BASEDIR_QUOTA_OVERIDE)); } long currentCount = content.getFileCount() + content.getDirectoryCount(); FileStatus[] executions = fs.listStatus(execHistory); Arrays.sort(executions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < executions.length - 1 && limit < currentCount; i++) { FileStatus stat = executions[i]; log.info("removing old execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } if (limit < currentCount) { FileStatus[] failedExecutions = fs.listStatus(execBasePath, new PathFilter() { public boolean accept(Path path) { try { dateFmt.parseDateTime(path.getName()); return true; } catch (IllegalArgumentException e) { return false; } } }); Arrays.sort(failedExecutions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < failedExecutions.length && limit < currentCount; i++) { FileStatus stat = failedExecutions[i]; log.info("removing failed execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } } if (executions.length > 0) { Path previous = executions[executions.length - 1].getPath(); FileInputFormat.setInputPaths(job, previous); log.info("Previous execution: " + previous.toString()); } else { System.out.println("No previous execution, all topics pulled from earliest available offset"); } DateTimeFormatter dateFmt = DateUtils.getDateTimeFormatter("YYYY-MM-dd-HH-mm-ss", DateTimeZone.UTC); String executionDate = new DateTime().toString(dateFmt); Path newExecutionOutput = new Path(execBasePath, executionDate); FileOutputFormat.setOutputPath(job, newExecutionOutput); log.info("New execution temp location: " + newExecutionOutput.toString()); EtlInputFormat.setLogger(log); job.setMapperClass(EtlMapper.class); job.setInputFormatClass(EtlInputFormat.class); job.setOutputFormatClass(EtlMultiOutputFormat.class); job.setNumReduceTasks(0); stopTiming("pre-setup"); job.submit(); job.waitForCompletion(true); Counters counters = job.getCounters(); for (String groupName : counters.getGroupNames()) { CounterGroup group = counters.getGroup(groupName); log.info("Group: " + group.getDisplayName()); for (Counter counter : group) { log.info(counter.getDisplayName() + ":\t" + counter.getValue()); } } stopTiming("hadoop"); startTiming("commit"); sendTrackingCounts(job, fs, newExecutionOutput); Map<EtlKey, ExceptionWritable> errors = readErrors(fs, newExecutionOutput); if (!errors.isEmpty()) log.error("Errors encountered during job run:"); for (Entry<EtlKey, ExceptionWritable> entry : errors.entrySet()) { log.error(entry.getKey().toString()); log.error(entry.getValue().toString()); } Path newHistory = new Path(execHistory, executionDate); log.info("Moving execution to history : " + newHistory); fs.rename(newExecutionOutput, newHistory); log.info("Job finished"); stopTiming("commit"); stopTiming("total"); createReport(job, timingMap); if (!job.isSuccessful()) { JobClient client = new JobClient(new JobConf(job.getConfiguration())); TaskCompletionEvent[] tasks = job.getTaskCompletionEvents(0); for (TaskReport task : client.getMapTaskReports(tasks[0].getTaskAttemptId().getJobID())) { if (task.getCurrentStatus().equals(TIPStatus.FAILED)) { for (String s : task.getDiagnostics()) { System.err.println("task error: " + s); } } } throw new RuntimeException("hadoop job failed"); } if (!errors.isEmpty() && props.getProperty(ETL_FAIL_ON_ERRORS, Boolean.FALSE.toString()).equalsIgnoreCase(Boolean.TRUE.toString())) { throw new RuntimeException("Camus saw errors, check stderr"); } }
CamusJob extends Configured implements Tool { public void run() throws Exception { startTiming("pre-setup"); startTiming("total"); Job job = createJob(props); if (getLog4jConfigure(job)) { DOMConfigurator.configure("log4j.xml"); } FileSystem fs = FileSystem.get(job.getConfiguration()); log.info("Dir Destination set to: " + EtlMultiOutputFormat.getDestinationPath(job)); Path execBasePath = new Path(props.getProperty(ETL_EXECUTION_BASE_PATH)); Path execHistory = new Path(props.getProperty(ETL_EXECUTION_HISTORY_PATH)); if (!fs.exists(execBasePath)) { log.info("The execution base path does not exist. Creating the directory"); fs.mkdirs(execBasePath); } if (!fs.exists(execHistory)) { log.info("The history base path does not exist. Creating the directory."); fs.mkdirs(execHistory); } ContentSummary content = fs.getContentSummary(execBasePath); long limit = (long) (content.getQuota() * job.getConfiguration().getFloat(ETL_EXECUTION_HISTORY_MAX_OF_QUOTA, (float) .5)); limit = limit == 0 ? 50000 : limit; if (props.containsKey(ETL_BASEDIR_QUOTA_OVERIDE)) { limit = Long.valueOf(props.getProperty(ETL_BASEDIR_QUOTA_OVERIDE)); } long currentCount = content.getFileCount() + content.getDirectoryCount(); FileStatus[] executions = fs.listStatus(execHistory); Arrays.sort(executions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < executions.length - 1 && limit < currentCount; i++) { FileStatus stat = executions[i]; log.info("removing old execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } if (limit < currentCount) { FileStatus[] failedExecutions = fs.listStatus(execBasePath, new PathFilter() { public boolean accept(Path path) { try { dateFmt.parseDateTime(path.getName()); return true; } catch (IllegalArgumentException e) { return false; } } }); Arrays.sort(failedExecutions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < failedExecutions.length && limit < currentCount; i++) { FileStatus stat = failedExecutions[i]; log.info("removing failed execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } } if (executions.length > 0) { Path previous = executions[executions.length - 1].getPath(); FileInputFormat.setInputPaths(job, previous); log.info("Previous execution: " + previous.toString()); } else { System.out.println("No previous execution, all topics pulled from earliest available offset"); } DateTimeFormatter dateFmt = DateUtils.getDateTimeFormatter("YYYY-MM-dd-HH-mm-ss", DateTimeZone.UTC); String executionDate = new DateTime().toString(dateFmt); Path newExecutionOutput = new Path(execBasePath, executionDate); FileOutputFormat.setOutputPath(job, newExecutionOutput); log.info("New execution temp location: " + newExecutionOutput.toString()); EtlInputFormat.setLogger(log); job.setMapperClass(EtlMapper.class); job.setInputFormatClass(EtlInputFormat.class); job.setOutputFormatClass(EtlMultiOutputFormat.class); job.setNumReduceTasks(0); stopTiming("pre-setup"); job.submit(); job.waitForCompletion(true); Counters counters = job.getCounters(); for (String groupName : counters.getGroupNames()) { CounterGroup group = counters.getGroup(groupName); log.info("Group: " + group.getDisplayName()); for (Counter counter : group) { log.info(counter.getDisplayName() + ":\t" + counter.getValue()); } } stopTiming("hadoop"); startTiming("commit"); sendTrackingCounts(job, fs, newExecutionOutput); Map<EtlKey, ExceptionWritable> errors = readErrors(fs, newExecutionOutput); if (!errors.isEmpty()) log.error("Errors encountered during job run:"); for (Entry<EtlKey, ExceptionWritable> entry : errors.entrySet()) { log.error(entry.getKey().toString()); log.error(entry.getValue().toString()); } Path newHistory = new Path(execHistory, executionDate); log.info("Moving execution to history : " + newHistory); fs.rename(newExecutionOutput, newHistory); log.info("Job finished"); stopTiming("commit"); stopTiming("total"); createReport(job, timingMap); if (!job.isSuccessful()) { JobClient client = new JobClient(new JobConf(job.getConfiguration())); TaskCompletionEvent[] tasks = job.getTaskCompletionEvents(0); for (TaskReport task : client.getMapTaskReports(tasks[0].getTaskAttemptId().getJobID())) { if (task.getCurrentStatus().equals(TIPStatus.FAILED)) { for (String s : task.getDiagnostics()) { System.err.println("task error: " + s); } } } throw new RuntimeException("hadoop job failed"); } if (!errors.isEmpty() && props.getProperty(ETL_FAIL_ON_ERRORS, Boolean.FALSE.toString()).equalsIgnoreCase(Boolean.TRUE.toString())) { throw new RuntimeException("Camus saw errors, check stderr"); } } }
CamusJob extends Configured implements Tool { public void run() throws Exception { startTiming("pre-setup"); startTiming("total"); Job job = createJob(props); if (getLog4jConfigure(job)) { DOMConfigurator.configure("log4j.xml"); } FileSystem fs = FileSystem.get(job.getConfiguration()); log.info("Dir Destination set to: " + EtlMultiOutputFormat.getDestinationPath(job)); Path execBasePath = new Path(props.getProperty(ETL_EXECUTION_BASE_PATH)); Path execHistory = new Path(props.getProperty(ETL_EXECUTION_HISTORY_PATH)); if (!fs.exists(execBasePath)) { log.info("The execution base path does not exist. Creating the directory"); fs.mkdirs(execBasePath); } if (!fs.exists(execHistory)) { log.info("The history base path does not exist. Creating the directory."); fs.mkdirs(execHistory); } ContentSummary content = fs.getContentSummary(execBasePath); long limit = (long) (content.getQuota() * job.getConfiguration().getFloat(ETL_EXECUTION_HISTORY_MAX_OF_QUOTA, (float) .5)); limit = limit == 0 ? 50000 : limit; if (props.containsKey(ETL_BASEDIR_QUOTA_OVERIDE)) { limit = Long.valueOf(props.getProperty(ETL_BASEDIR_QUOTA_OVERIDE)); } long currentCount = content.getFileCount() + content.getDirectoryCount(); FileStatus[] executions = fs.listStatus(execHistory); Arrays.sort(executions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < executions.length - 1 && limit < currentCount; i++) { FileStatus stat = executions[i]; log.info("removing old execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } if (limit < currentCount) { FileStatus[] failedExecutions = fs.listStatus(execBasePath, new PathFilter() { public boolean accept(Path path) { try { dateFmt.parseDateTime(path.getName()); return true; } catch (IllegalArgumentException e) { return false; } } }); Arrays.sort(failedExecutions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < failedExecutions.length && limit < currentCount; i++) { FileStatus stat = failedExecutions[i]; log.info("removing failed execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } } if (executions.length > 0) { Path previous = executions[executions.length - 1].getPath(); FileInputFormat.setInputPaths(job, previous); log.info("Previous execution: " + previous.toString()); } else { System.out.println("No previous execution, all topics pulled from earliest available offset"); } DateTimeFormatter dateFmt = DateUtils.getDateTimeFormatter("YYYY-MM-dd-HH-mm-ss", DateTimeZone.UTC); String executionDate = new DateTime().toString(dateFmt); Path newExecutionOutput = new Path(execBasePath, executionDate); FileOutputFormat.setOutputPath(job, newExecutionOutput); log.info("New execution temp location: " + newExecutionOutput.toString()); EtlInputFormat.setLogger(log); job.setMapperClass(EtlMapper.class); job.setInputFormatClass(EtlInputFormat.class); job.setOutputFormatClass(EtlMultiOutputFormat.class); job.setNumReduceTasks(0); stopTiming("pre-setup"); job.submit(); job.waitForCompletion(true); Counters counters = job.getCounters(); for (String groupName : counters.getGroupNames()) { CounterGroup group = counters.getGroup(groupName); log.info("Group: " + group.getDisplayName()); for (Counter counter : group) { log.info(counter.getDisplayName() + ":\t" + counter.getValue()); } } stopTiming("hadoop"); startTiming("commit"); sendTrackingCounts(job, fs, newExecutionOutput); Map<EtlKey, ExceptionWritable> errors = readErrors(fs, newExecutionOutput); if (!errors.isEmpty()) log.error("Errors encountered during job run:"); for (Entry<EtlKey, ExceptionWritable> entry : errors.entrySet()) { log.error(entry.getKey().toString()); log.error(entry.getValue().toString()); } Path newHistory = new Path(execHistory, executionDate); log.info("Moving execution to history : " + newHistory); fs.rename(newExecutionOutput, newHistory); log.info("Job finished"); stopTiming("commit"); stopTiming("total"); createReport(job, timingMap); if (!job.isSuccessful()) { JobClient client = new JobClient(new JobConf(job.getConfiguration())); TaskCompletionEvent[] tasks = job.getTaskCompletionEvents(0); for (TaskReport task : client.getMapTaskReports(tasks[0].getTaskAttemptId().getJobID())) { if (task.getCurrentStatus().equals(TIPStatus.FAILED)) { for (String s : task.getDiagnostics()) { System.err.println("task error: " + s); } } } throw new RuntimeException("hadoop job failed"); } if (!errors.isEmpty() && props.getProperty(ETL_FAIL_ON_ERRORS, Boolean.FALSE.toString()).equalsIgnoreCase(Boolean.TRUE.toString())) { throw new RuntimeException("Camus saw errors, check stderr"); } } CamusJob();  CamusJob(Properties props);  CamusJob(Properties props, Logger log); }
CamusJob extends Configured implements Tool { public void run() throws Exception { startTiming("pre-setup"); startTiming("total"); Job job = createJob(props); if (getLog4jConfigure(job)) { DOMConfigurator.configure("log4j.xml"); } FileSystem fs = FileSystem.get(job.getConfiguration()); log.info("Dir Destination set to: " + EtlMultiOutputFormat.getDestinationPath(job)); Path execBasePath = new Path(props.getProperty(ETL_EXECUTION_BASE_PATH)); Path execHistory = new Path(props.getProperty(ETL_EXECUTION_HISTORY_PATH)); if (!fs.exists(execBasePath)) { log.info("The execution base path does not exist. Creating the directory"); fs.mkdirs(execBasePath); } if (!fs.exists(execHistory)) { log.info("The history base path does not exist. Creating the directory."); fs.mkdirs(execHistory); } ContentSummary content = fs.getContentSummary(execBasePath); long limit = (long) (content.getQuota() * job.getConfiguration().getFloat(ETL_EXECUTION_HISTORY_MAX_OF_QUOTA, (float) .5)); limit = limit == 0 ? 50000 : limit; if (props.containsKey(ETL_BASEDIR_QUOTA_OVERIDE)) { limit = Long.valueOf(props.getProperty(ETL_BASEDIR_QUOTA_OVERIDE)); } long currentCount = content.getFileCount() + content.getDirectoryCount(); FileStatus[] executions = fs.listStatus(execHistory); Arrays.sort(executions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < executions.length - 1 && limit < currentCount; i++) { FileStatus stat = executions[i]; log.info("removing old execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } if (limit < currentCount) { FileStatus[] failedExecutions = fs.listStatus(execBasePath, new PathFilter() { public boolean accept(Path path) { try { dateFmt.parseDateTime(path.getName()); return true; } catch (IllegalArgumentException e) { return false; } } }); Arrays.sort(failedExecutions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < failedExecutions.length && limit < currentCount; i++) { FileStatus stat = failedExecutions[i]; log.info("removing failed execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } } if (executions.length > 0) { Path previous = executions[executions.length - 1].getPath(); FileInputFormat.setInputPaths(job, previous); log.info("Previous execution: " + previous.toString()); } else { System.out.println("No previous execution, all topics pulled from earliest available offset"); } DateTimeFormatter dateFmt = DateUtils.getDateTimeFormatter("YYYY-MM-dd-HH-mm-ss", DateTimeZone.UTC); String executionDate = new DateTime().toString(dateFmt); Path newExecutionOutput = new Path(execBasePath, executionDate); FileOutputFormat.setOutputPath(job, newExecutionOutput); log.info("New execution temp location: " + newExecutionOutput.toString()); EtlInputFormat.setLogger(log); job.setMapperClass(EtlMapper.class); job.setInputFormatClass(EtlInputFormat.class); job.setOutputFormatClass(EtlMultiOutputFormat.class); job.setNumReduceTasks(0); stopTiming("pre-setup"); job.submit(); job.waitForCompletion(true); Counters counters = job.getCounters(); for (String groupName : counters.getGroupNames()) { CounterGroup group = counters.getGroup(groupName); log.info("Group: " + group.getDisplayName()); for (Counter counter : group) { log.info(counter.getDisplayName() + ":\t" + counter.getValue()); } } stopTiming("hadoop"); startTiming("commit"); sendTrackingCounts(job, fs, newExecutionOutput); Map<EtlKey, ExceptionWritable> errors = readErrors(fs, newExecutionOutput); if (!errors.isEmpty()) log.error("Errors encountered during job run:"); for (Entry<EtlKey, ExceptionWritable> entry : errors.entrySet()) { log.error(entry.getKey().toString()); log.error(entry.getValue().toString()); } Path newHistory = new Path(execHistory, executionDate); log.info("Moving execution to history : " + newHistory); fs.rename(newExecutionOutput, newHistory); log.info("Job finished"); stopTiming("commit"); stopTiming("total"); createReport(job, timingMap); if (!job.isSuccessful()) { JobClient client = new JobClient(new JobConf(job.getConfiguration())); TaskCompletionEvent[] tasks = job.getTaskCompletionEvents(0); for (TaskReport task : client.getMapTaskReports(tasks[0].getTaskAttemptId().getJobID())) { if (task.getCurrentStatus().equals(TIPStatus.FAILED)) { for (String s : task.getDiagnostics()) { System.err.println("task error: " + s); } } } throw new RuntimeException("hadoop job failed"); } if (!errors.isEmpty() && props.getProperty(ETL_FAIL_ON_ERRORS, Boolean.FALSE.toString()).equalsIgnoreCase(Boolean.TRUE.toString())) { throw new RuntimeException("Camus saw errors, check stderr"); } } CamusJob();  CamusJob(Properties props);  CamusJob(Properties props, Logger log); static void startTiming(String name); static void stopTiming(String name); static void setTime(String name); static void populateConf(Properties props, Configuration conf, Logger log); void run(); void cancel(); Map<EtlKey, ExceptionWritable> readErrors(FileSystem fs, Path newExecutionOutput); void sendTrackingCounts(JobContext job, FileSystem fs, Path newExecutionOutput); static void main(String[] args); @SuppressWarnings("static-access") @Override int run(String[] args); static boolean getPostTrackingCountsToKafka(Job job); static int getKafkaFetchRequestMinBytes(JobContext context); static int getKafkaFetchRequestMaxWait(JobContext job); static String getKafkaBrokers(JobContext job); static int getKafkaFetchRequestCorrelationId(JobContext job); static String getKafkaClientName(JobContext job); static String getKafkaFetchRequestBufferSize(JobContext job); static int getKafkaTimeoutValue(JobContext job); static int getKafkaBufferSize(JobContext job); static boolean getLog4jConfigure(JobContext job); static String getReporterClass(JobContext job); }
CamusJob extends Configured implements Tool { public void run() throws Exception { startTiming("pre-setup"); startTiming("total"); Job job = createJob(props); if (getLog4jConfigure(job)) { DOMConfigurator.configure("log4j.xml"); } FileSystem fs = FileSystem.get(job.getConfiguration()); log.info("Dir Destination set to: " + EtlMultiOutputFormat.getDestinationPath(job)); Path execBasePath = new Path(props.getProperty(ETL_EXECUTION_BASE_PATH)); Path execHistory = new Path(props.getProperty(ETL_EXECUTION_HISTORY_PATH)); if (!fs.exists(execBasePath)) { log.info("The execution base path does not exist. Creating the directory"); fs.mkdirs(execBasePath); } if (!fs.exists(execHistory)) { log.info("The history base path does not exist. Creating the directory."); fs.mkdirs(execHistory); } ContentSummary content = fs.getContentSummary(execBasePath); long limit = (long) (content.getQuota() * job.getConfiguration().getFloat(ETL_EXECUTION_HISTORY_MAX_OF_QUOTA, (float) .5)); limit = limit == 0 ? 50000 : limit; if (props.containsKey(ETL_BASEDIR_QUOTA_OVERIDE)) { limit = Long.valueOf(props.getProperty(ETL_BASEDIR_QUOTA_OVERIDE)); } long currentCount = content.getFileCount() + content.getDirectoryCount(); FileStatus[] executions = fs.listStatus(execHistory); Arrays.sort(executions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < executions.length - 1 && limit < currentCount; i++) { FileStatus stat = executions[i]; log.info("removing old execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } if (limit < currentCount) { FileStatus[] failedExecutions = fs.listStatus(execBasePath, new PathFilter() { public boolean accept(Path path) { try { dateFmt.parseDateTime(path.getName()); return true; } catch (IllegalArgumentException e) { return false; } } }); Arrays.sort(failedExecutions, new Comparator<FileStatus>() { public int compare(FileStatus f1, FileStatus f2) { return f1.getPath().getName().compareTo(f2.getPath().getName()); } }); for (int i = 0; i < failedExecutions.length && limit < currentCount; i++) { FileStatus stat = failedExecutions[i]; log.info("removing failed execution: " + stat.getPath().getName()); ContentSummary execContent = fs.getContentSummary(stat.getPath()); currentCount -= execContent.getFileCount() + execContent.getDirectoryCount(); fs.delete(stat.getPath(), true); } } if (executions.length > 0) { Path previous = executions[executions.length - 1].getPath(); FileInputFormat.setInputPaths(job, previous); log.info("Previous execution: " + previous.toString()); } else { System.out.println("No previous execution, all topics pulled from earliest available offset"); } DateTimeFormatter dateFmt = DateUtils.getDateTimeFormatter("YYYY-MM-dd-HH-mm-ss", DateTimeZone.UTC); String executionDate = new DateTime().toString(dateFmt); Path newExecutionOutput = new Path(execBasePath, executionDate); FileOutputFormat.setOutputPath(job, newExecutionOutput); log.info("New execution temp location: " + newExecutionOutput.toString()); EtlInputFormat.setLogger(log); job.setMapperClass(EtlMapper.class); job.setInputFormatClass(EtlInputFormat.class); job.setOutputFormatClass(EtlMultiOutputFormat.class); job.setNumReduceTasks(0); stopTiming("pre-setup"); job.submit(); job.waitForCompletion(true); Counters counters = job.getCounters(); for (String groupName : counters.getGroupNames()) { CounterGroup group = counters.getGroup(groupName); log.info("Group: " + group.getDisplayName()); for (Counter counter : group) { log.info(counter.getDisplayName() + ":\t" + counter.getValue()); } } stopTiming("hadoop"); startTiming("commit"); sendTrackingCounts(job, fs, newExecutionOutput); Map<EtlKey, ExceptionWritable> errors = readErrors(fs, newExecutionOutput); if (!errors.isEmpty()) log.error("Errors encountered during job run:"); for (Entry<EtlKey, ExceptionWritable> entry : errors.entrySet()) { log.error(entry.getKey().toString()); log.error(entry.getValue().toString()); } Path newHistory = new Path(execHistory, executionDate); log.info("Moving execution to history : " + newHistory); fs.rename(newExecutionOutput, newHistory); log.info("Job finished"); stopTiming("commit"); stopTiming("total"); createReport(job, timingMap); if (!job.isSuccessful()) { JobClient client = new JobClient(new JobConf(job.getConfiguration())); TaskCompletionEvent[] tasks = job.getTaskCompletionEvents(0); for (TaskReport task : client.getMapTaskReports(tasks[0].getTaskAttemptId().getJobID())) { if (task.getCurrentStatus().equals(TIPStatus.FAILED)) { for (String s : task.getDiagnostics()) { System.err.println("task error: " + s); } } } throw new RuntimeException("hadoop job failed"); } if (!errors.isEmpty() && props.getProperty(ETL_FAIL_ON_ERRORS, Boolean.FALSE.toString()).equalsIgnoreCase(Boolean.TRUE.toString())) { throw new RuntimeException("Camus saw errors, check stderr"); } } CamusJob();  CamusJob(Properties props);  CamusJob(Properties props, Logger log); static void startTiming(String name); static void stopTiming(String name); static void setTime(String name); static void populateConf(Properties props, Configuration conf, Logger log); void run(); void cancel(); Map<EtlKey, ExceptionWritable> readErrors(FileSystem fs, Path newExecutionOutput); void sendTrackingCounts(JobContext job, FileSystem fs, Path newExecutionOutput); static void main(String[] args); @SuppressWarnings("static-access") @Override int run(String[] args); static boolean getPostTrackingCountsToKafka(Job job); static int getKafkaFetchRequestMinBytes(JobContext context); static int getKafkaFetchRequestMaxWait(JobContext job); static String getKafkaBrokers(JobContext job); static int getKafkaFetchRequestCorrelationId(JobContext job); static String getKafkaClientName(JobContext job); static String getKafkaFetchRequestBufferSize(JobContext job); static int getKafkaTimeoutValue(JobContext job); static int getKafkaBufferSize(JobContext job); static boolean getLog4jConfigure(JobContext job); static String getReporterClass(JobContext job); static final String ETL_EXECUTION_BASE_PATH; static final String ETL_EXECUTION_HISTORY_PATH; static final String ETL_COUNTS_PATH; static final String ETL_KEEP_COUNT_FILES; static final String ETL_BASEDIR_QUOTA_OVERIDE; static final String ETL_EXECUTION_HISTORY_MAX_OF_QUOTA; static final String ETL_FAIL_ON_ERRORS; static final String ZK_AUDIT_HOSTS; static final String KAFKA_MONITOR_TIER; static final String CAMUS_MESSAGE_ENCODER_CLASS; static final String BROKER_URI_FILE; static final String POST_TRACKING_COUNTS_TO_KAFKA; static final String KAFKA_FETCH_REQUEST_MAX_WAIT; static final String KAFKA_FETCH_REQUEST_MIN_BYTES; static final String KAFKA_FETCH_REQUEST_CORRELATION_ID; static final String KAFKA_CLIENT_NAME; static final String KAFKA_FETCH_BUFFER_SIZE; static final String KAFKA_BROKERS; static final String KAFKA_HOST_URL; static final String KAFKA_HOST_PORT; static final String KAFKA_TIMEOUT_VALUE; static final String CAMUS_REPORTER_CLASS; static final String LOG4J_CONFIGURATION; }