@Override
  public int getPartition(Object key) {
    final Pair<HoodieFileGroupId, String> parts = (Pair<HoodieFileGroupId, String>) key;
    // TODO replace w/ more performant hash
    final long hashOfKey = NumericUtils.getMessageDigestHash("MD5", parts.getRight());
    final List<Integer> candidatePartitions = fileGroupToPartitions.get(parts.getLeft());
    final int idx = (int) Math.floorMod((int) hashOfKey, candidatePartitions.size());
    assert idx >= 0;
    return candidatePartitions.get(idx);
  }
@Test
  public void testGetPartitions() {
    Map<HoodieFileGroupId, Long> comparisons1 = new HashMap<HoodieFileGroupId, Long>() {
      {
        IntStream.range(0, 100000).forEach(f -> put(new HoodieFileGroupId("p1", "f" + f), 100L));
      }
    };
    BucketizedBloomCheckPartitioner p = new BucketizedBloomCheckPartitioner(1000, comparisons1, 10);

    IntStream.range(0, 100000).forEach(f -> {
      int partition = p.getPartition(Pair.of(new HoodieFileGroupId("p1", "f" + f), "value"));
      assertTrue(0 <= partition && partition <= 1000, "partition is out of range: " + partition);
    });
  }