MetadataExtractorMapper extends Mapper<AvroKey<DocumentContent>, NullWritable, NullWritable, NullWritable> { @Override protected void setup(Context context) throws IOException, InterruptedException { namedOutputMeta = context.getConfiguration().get(NAMED_OUTPUT_META); if (namedOutputMeta == null || namedOutputMeta.isEmpty()) { throw new RuntimeException("no named output provided for metadata"); } namedOutputFault = context.getConfiguration().get(NAMED_OUTPUT_FAULT); if (namedOutputFault == null || namedOutputFault.isEmpty()) { throw new RuntimeException("no named output provided for fault"); } String excludedIdsCSV = context.getConfiguration().get(EXCLUDED_IDS); if (excludedIdsCSV != null && !excludedIdsCSV.trim().isEmpty() && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(excludedIdsCSV)) { log.info("got excluded ids: " + excludedIdsCSV); excludedIds = new HashSet<String>(Arrays.asList(StringUtils.split(excludedIdsCSV.trim(), ','))); } else { log.info("got no excluded ids"); } interruptionTimeoutSecs = WorkflowRuntimeParameters.getIntegerParamValue( INTERRUPT_PROCESSING_TIME_THRESHOLD_SECS, context.getConfiguration()); Integer processingTimeThresholdSecs = WorkflowRuntimeParameters.getIntegerParamValue( LOG_FAULT_PROCESSING_TIME_THRESHOLD_SECS, context.getConfiguration()); if (processingTimeThresholdSecs != null) { this.processingTimeThreshold = SECS_TO_MILLIS * processingTimeThresholdSecs; } Counter invalidPdfCounter = context.getCounter(InvalidRecordCounters.INVALID_PDF_HEADER); invalidPdfCounter.setValue(0); this.contentApprover = new InvalidCountableContentApproverWrapper(new PDFHeaderBasedContentApprover(), invalidPdfCounter); mos = instantiateMultipleOutputs(context); currentProgress = 0; intervalTime = System.currentTimeMillis(); }  @Override void cleanup(Context context); @Override void map(AvroKey<DocumentContent> key, NullWritable ignore, Context context); static final String NAMED_OUTPUT_META; static final String NAMED_OUTPUT_FAULT; static final String EXCLUDED_IDS; static final String LOG_FAULT_PROCESSING_TIME_THRESHOLD_SECS; static final String INTERRUPT_PROCESSING_TIME_THRESHOLD_SECS; static final String FAULT_CODE_PROCESSING_TIME_THRESHOLD_EXCEEDED; static final String FAULT_SUPPLEMENTARY_DATA_PROCESSING_TIME; }
@Test(expected=RuntimeException.class) public void testSetupWithoutNamedOutputMeta() throws Exception { Configuration conf = new Configuration(); doReturn(conf).when(context).getConfiguration(); mapper.setup(context); }
@Test(expected=RuntimeException.class) public void testSetupWithoutNamedOutputFault() throws Exception { Configuration conf = new Configuration(); conf.set(NAMED_OUTPUT_META, "meta"); doReturn(conf).when(context).getConfiguration(); mapper.setup(context); }