MetadataExtractorMapper extends Mapper<AvroKey<DocumentContent>, NullWritable, NullWritable, NullWritable> { @Override public void map(AvroKey<DocumentContent> key, NullWritable ignore, Context context) throws IOException, InterruptedException { DocumentContent content = key.datum(); String documentId = content.getId().toString(); if (excludedIds.contains(documentId)) { log.info("skipping processing for excluded id " + documentId); return; } if (content.getPdf()!=null) { ByteBuffer byteBuffer = content.getPdf(); if (byteBuffer.hasArray() && contentApprover.approve(byteBuffer.array())) { try (InputStream inputStream = new ByteBufferInputStream(byteBuffer)) { processStream(documentId, inputStream); } } else { log.info(invalidPdfHeaderMsg); handleException(new InvalidPdfException(invalidPdfHeaderMsg), content.getId().toString()); } } else { log.warn("no byte data found for id: " + content.getId()); } }  @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 public void testMap() throws Exception { Configuration conf = new Configuration(); conf.set(NAMED_OUTPUT_META, "meta"); conf.set(NAMED_OUTPUT_FAULT, "fault"); doReturn(conf).when(context).getConfiguration(); doReturn(invalidPdfCounter).when(context).getCounter(INVALID_PDF_HEADER); mapper.setup(context); String id = "id"; DocumentContent.Builder docContentBuilder = DocumentContent.newBuilder(); docContentBuilder.setId(id); docContentBuilder.setPdf(ByteBuffer.wrap(getContent(PDF_FILE))); mapper.map(new AvroKey<>(docContentBuilder.build()), null, context); verify(context, never()).write(any(), any()); verify(multipleOutputs, times(1)).write(mosKeyCaptor.capture(), mosValueCaptor.capture()); assertEquals(conf.get(NAMED_OUTPUT_META), mosKeyCaptor.getValue()); ExtractedDocumentMetadata docMeta = (ExtractedDocumentMetadata) mosValueCaptor.getValue().datum(); assertNotNull(docMeta); assertEquals(id, docMeta.getId()); verify(invalidPdfCounter, never()).increment(1); }