ExternalIdReferenceExtractor implements Serializable { public JavaPairRDD<String, Citation> extractExternalIdReferences(JavaRDD<DocumentMetadata> documentsMetadata, String idType) { Preconditions.checkNotNull(documentsMetadata); Preconditions.checkNotNull(idType); return documentsMetadata .flatMapToPair(metadata -> { List<Tuple2<String, Citation>> externalIdReferences = Lists.newArrayList(); for (ReferenceMetadata referenceMetadata : metadata.getReferences()) { if (referenceMetadata.getExternalIds() == null || !referenceMetadata.getExternalIds().containsKey(idType)) { continue; } String externalId = referenceMetadata.getExternalIds().get(idType).toString(); Citation partialCitation = new Citation(metadata.getId(), referenceMetadata.getPosition(), null); externalIdReferences.add(new Tuple2<>(externalId, partialCitation)); } return externalIdReferences.iterator(); }); }  JavaPairRDD<String, Citation> extractExternalIdReferences(JavaRDD<DocumentMetadata> documentsMetadata, String idType);  }
@Test(expected = NullPointerException.class) public void extractExternalIdReferences_NULL_ID_TYPE() { externalIdReferenceExtractor.extractExternalIdReferences(docMetadataRdd, null); }
@Test(expected = NullPointerException.class) public void extractExternalIdReferences_NULL_DOC_METADATA() { externalIdReferenceExtractor.extractExternalIdReferences(null, "someIdType"); }
@Test public void extractExternalIdReferences() throws Exception { doReturn(externalIdReferencesRdd).when(docMetadataRdd).flatMapToPair(any()); JavaPairRDD<String, Citation> retExternalIdReferencesRdd = externalIdReferenceExtractor.extractExternalIdReferences(docMetadataRdd, "someIdType"); assertSame(retExternalIdReferencesRdd, externalIdReferencesRdd); verify(docMetadataRdd).flatMapToPair(flatMapCitationsFunctionArg.capture()); assertFlatMapCitationsFunction(flatMapCitationsFunctionArg.getValue()); }