IisAffiliationReader implements Serializable, AffiliationReader { @Override public JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath) { Preconditions.checkNotNull(sc); Preconditions.checkArgument(StringUtils.isNotBlank(inputPath)); JavaRDD<ExtractedDocumentMetadata> sourceAffiliations = sparkAvroLoader .loadJavaRDD(sc, inputPath, ExtractedDocumentMetadata.class); return sourceAffiliations.flatMap(srcAff -> affiliationConverter.convert(srcAff).iterator()); }  @Override JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath);  }
@Test(expected = NullPointerException.class) public void readAffiliations_sparkContext_null() { reader.readAffiliations(null, "/aaa"); }
@Test(expected = IllegalArgumentException.class) public void readAffiliations_inputPath_blank() { reader.readAffiliations(sparkContext, " "); }
@Test public void readAffiliations() throws Exception { String inputPath = "/data/affiliations"; when(sparkAvroLoader.loadJavaRDD(sparkContext, inputPath, ExtractedDocumentMetadata.class)).thenReturn(inputDocuments); doReturn(affMatchAffiliations).when(inputDocuments).flatMap(any()); JavaRDD<AffMatchAffiliation> retAffMatchAffiliations = reader.readAffiliations(sparkContext, inputPath); assertTrue(affMatchAffiliations == retAffMatchAffiliations); verify(inputDocuments).flatMap(convertFunction.capture()); assertConvertFunction(convertFunction.getValue()); }