WriteInfoPartitioner { @VisibleForTesting Map<PartitionInfo, List<WriteInfo>> partitionWritesByShardStrategyTimestamp(List<WriteInfo> writes) { int shards = numShards.get(); return writes.stream() .collect(Collectors.groupingBy(write -> getPartitionInfo(write, shards), Collectors.toList())); } WriteInfoPartitioner(KeyValueService kvs, Supplier<Integer> numShards); Map<PartitionInfo, List<WriteInfo>> filterAndPartition(List<WriteInfo> writes);  }
@Test public void partitionWritesByShardStrategyTimestampPartitionsIntoSeparatePartitions() { List<WriteInfo> writes = ImmutableList.of( getWriteInfo(CONSERVATIVE, 0, 0, 100L), getWriteInfo(CONSERVATIVE, 1, 0, 100L), getWriteInfo(CONSERVATIVE, 0, 3, 100L), getWriteInfo(CONSERVATIVE, 0, 0, 200L), getWriteInfo(CONSERVATIVE2, 0, 0, 100L), getWriteInfo(THOROUGH, 0, 0, 100L)); Map<PartitionInfo, List<WriteInfo>> partitions = partitioner.partitionWritesByShardStrategyTimestamp(writes); assertThat(partitions.size()).isEqualTo(6); }
@Test public void partitionWritesByShardStrategyTimestampGroupsOnShardClash() { List<WriteInfo> writes = new ArrayList<>(); for (int i = 0; i <= numShards; i++) { writes.add(getWriteInfoWithFixedCellHash(CONSERVATIVE, i)); } Map<PartitionInfo, List<WriteInfo>> partitions = partitioner.partitionWritesByShardStrategyTimestamp(writes); assertThat(partitions.keySet()) .containsExactly(PartitionInfo.of(writes.get(0).toShard(numShards), true, 1L)); assertThat(Iterables.getOnlyElement(partitions.values())).isEqualTo(writes); }