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