@Test
public void attachCitationsAndLimitDocs() {

    
    
    // given
    
    JavaPairRDD<String, MatchableEntity> citIdDocPairs = sparkContext.parallelizePairs(ImmutableList.of(
            new Tuple2<>(citation1.id(), document1),
            new Tuple2<>(citation1.id(), document2),
            new Tuple2<>(citation2.id(), document3)));
    
    JavaPairRDD<String, MatchableEntity> citations = sparkContext.parallelizePairs(generateIdWithEntityTuples(
            Lists.newArrayList(citation1, citation2, citation3, citation4, citation5)));
    
    
    
    // execute
    
    JavaPairRDD<MatchableEntity, MatchableEntity> actualCitDocPairs = citationAttacher.attachCitationsAndLimitDocs(citIdDocPairs, citations);
    
    
    // assert
    List<Tuple2<MatchableEntity, MatchableEntity>> expectedCitDocPairs = ImmutableList.of(
            new Tuple2<>(citation1, document1),
            new Tuple2<>(citation1, document2),
            new Tuple2<>(citation2, document3));
    
    assertCitDocPairsEquals(actualCitDocPairs.collect(), expectedCitDocPairs);
    
}