@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); }
@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 { @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 { @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); }
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); }
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);  }