ConfigurableCitationMatchingService { public void matchCitations(JavaSparkContext sparkContext, String inputCitationPath, String inputDocumentPath, String outputPath) { JavaPairRDD<INPUT_CIT_KEY, INPUT_CIT_VALUE> citations = inputCitationReader.readCitations(sparkContext, inputCitationPath); JavaPairRDD<String, MatchableEntity> citationsConverted = inputCitationConverter.convertCitations(citations); citationsConverted = citationsConverted.partitionBy(new HashPartitioner(numberOfPartitions)); JavaPairRDD<INPUT_DOC_KEY, INPUT_DOC_VALUE> documents = inputDocumentReader.readDocuments(sparkContext, inputDocumentPath); JavaPairRDD<String, MatchableEntity> documentsConverted = inputDocumentConverter.convertDocuments(documents); documentsConverted = documentsConverted.partitionBy(new HashPartitioner(numberOfPartitions)); JavaPairRDD<MatchableEntity, IdWithSimilarity> matchedCitations = coreCitationMatchingService.matchCitations(citationsConverted, documentsConverted); JavaPairRDD<OUTPUT_MATCHED_KEY, OUTPUT_MACHED_VALUE> matchedCitationsConverted = outputConverter.convertMatchedCitations(matchedCitations); outputWriter.writeMatchedCitations(matchedCitationsConverted, outputPath); }  void matchCitations(JavaSparkContext sparkContext, String inputCitationPath, String inputDocumentPath, String outputPath); void setInputCitationReader(InputCitationReader<INPUT_CIT_KEY, INPUT_CIT_VALUE> inputCitationReader); void setInputCitationConverter(InputCitationConverter<INPUT_CIT_KEY, INPUT_CIT_VALUE> inputCitationConverter); void setInputDocumentReader(InputDocumentReader<INPUT_DOC_KEY, INPUT_DOC_VALUE> inputDocumentReader); void setInputDocumentConverter(InputDocumentConverter<INPUT_DOC_KEY, INPUT_DOC_VALUE> inputDocumentConverter); void setOutputConverter(OutputConverter<OUTPUT_MATCHED_KEY, OUTPUT_MACHED_VALUE> outputConverter); void setOutputWriter(OutputWriter<OUTPUT_MATCHED_KEY, OUTPUT_MACHED_VALUE> outputWriter); void setCoreCitationMatchingService(CoreCitationMatchingService coreCitationMatchingService); void setNumberOfPartitions(Integer numberOfPartitions);  }
@Test public void matchCitations() { when(inputCitationReader.readCitations(sparkContext, "/input/cit/path")).thenReturn(citations); when(inputCitationConverter.convertCitations(citations)).thenReturn(convertedCitations); when(convertedCitations.partitions()).thenReturn(Lists.newArrayList(16)); when(convertedCitations.partitionBy(any())).thenReturn(repartitionedCitations); when(inputDocumentReader.readDocuments(sparkContext, "/input/doc/path")).thenReturn(documents); when(inputDocumentConverter.convertDocuments(documents)).thenReturn(convertedDocuments); when(convertedDocuments.partitions()).thenReturn(Lists.newArrayList(12)); when(convertedDocuments.partitionBy(any())).thenReturn(repartitionedDocuments); when(coreCitationMatchingService.matchCitations(repartitionedCitations, repartitionedDocuments)).thenReturn(matched); when(outputConverter.convertMatchedCitations(matched)).thenReturn(convertedMatched); citationMatchingService.matchCitations(sparkContext, "/input/cit/path", "/input/doc/path", "/output/path"); verify(inputCitationReader).readCitations(sparkContext, "/input/cit/path"); verify(inputCitationConverter).convertCitations(citations); verify(convertedCitations).partitionBy(citationsPartitioner.capture()); assertPartitioner(citationsPartitioner.getValue(), 5); verify(inputDocumentReader).readDocuments(sparkContext, "/input/doc/path"); verify(inputDocumentConverter).convertDocuments(documents); verify(convertedDocuments).partitionBy(documentsPartitioner.capture()); assertPartitioner(documentsPartitioner.getValue(), 5); verify(coreCitationMatchingService).matchCitations(repartitionedCitations, repartitionedDocuments); verify(outputConverter).convertMatchedCitations(matched); verify(outputWriter).writeMatchedCitations(convertedMatched, "/output/path"); }