@Test
  public void extractInvalidHashes() {

      // given
      
      SparkConf conf = new SparkConf().setMaster("local[1]").setAppName("HashBucketLimiterTest");
      
      try (JavaSparkContext sc = new JavaSparkContext(conf)) {
          
          @SuppressWarnings("unchecked")
          List<Tuple2<String, String>> hashes1 = Lists.newArrayList(tuple("1", "x"), tuple("1", "y"), tuple("3", "z"), tuple("1", "x"), tuple("2", "x"), tuple("5", "z"));
          @SuppressWarnings("unchecked")
          List<Tuple2<String, String>> hashes2 = Lists.newArrayList(tuple("1", "x"), tuple("1", "a"), tuple("3", "z"), tuple("3", "aa"), tuple("4", "a"), tuple("5", "z"));
          
          JavaPairRDD<String, String> hash1Rdd = sc.parallelizePairs(hashes1);
          JavaPairRDD<String, String> hash2Rdd = sc.parallelizePairs(hashes2);
          
          // execute
          
          JavaPairRDD<String, Long> invalidHashes = invalidHashExtractor.extractInvalidHashes(hash1Rdd, hash2Rdd, 2);
          
          
          // assert
          
          assertEquals(3, invalidHashes.count());
          checkContains(invalidHashes, "1", 6);
          checkContains(invalidHashes, "2", 0);
          checkContains(invalidHashes, "4", 0);
      }
}