CoansysInputCitationReader implements InputCitationReader<String, ReferenceMetadata>, Serializable { @Override public JavaPairRDD<String, ReferenceMetadata> readCitations(JavaSparkContext sparkContext, String inputCitationPath) { JavaPairRDD<Writable, BytesWritable> rawCitations = sparkContext.sequenceFile(inputCitationPath, Writable.class, BytesWritable.class); JavaRDD<DocumentWrapper> docWrappers = rawCitations.map(bw -> bytesWritableConverter.convertToDocumentWrapper(bw._2())); final int maxSupportedCitationLength = this.maxSupportedCitationLength; JavaPairRDD<String, ReferenceMetadata> docReferences = docWrappers.flatMapToPair(docWrapper -> referenceExtractor.extractReferences(docWrapper, maxSupportedCitationLength)); return docReferences; }  @Override JavaPairRDD<String, ReferenceMetadata> readCitations(JavaSparkContext sparkContext, String inputCitationPath); void setBytesWritableConverter(BytesWritableConverter bytesWritableConverter); void setReferenceExtractor(ReferenceExtractor referenceExtractor); void setMaxSupportedCitationLength(int maxSupportedCitationLength);  }
@Test public void readCitations() throws Exception { String inputPath = "/this/is/a/path"; when(sparkContext.sequenceFile(inputPath, Writable.class, BytesWritable.class)).thenReturn(rawCitations); doReturn(docWrappers).when(rawCitations).map(Mockito.any()); doReturn(docReferences).when(docWrappers).flatMapToPair(Mockito.any()); JavaPairRDD<String, ReferenceMetadata> citReferences = coansysInputCitationReader.readCitations(sparkContext, inputPath); assertTrue(docReferences == citReferences); verify(sparkContext).sequenceFile(inputPath, Writable.class, BytesWritable.class); verify(rawCitations).map(convertMatchedCitationFunction.capture()); assertConvertToDocumentWrapperFunction(convertMatchedCitationFunction.getValue()); verify(docWrappers).flatMapToPair(extractReferencesFunction.capture()); assertExtractReferencesFunction(extractReferencesFunction.getValue()); }