GroupBySystemStreamPartition implements SystemStreamPartitionGrouper { @Override public Map<TaskName, Set<SystemStreamPartition>> group(Set<SystemStreamPartition> ssps) { Map<TaskName, Set<SystemStreamPartition>> groupedMap = new HashMap<TaskName, Set<SystemStreamPartition>>(); for (SystemStreamPartition ssp : ssps) { if (broadcastStreams.contains(ssp)) { continue; } HashSet<SystemStreamPartition> sspSet = new HashSet<SystemStreamPartition>(); sspSet.add(ssp); groupedMap.put(new TaskName(ssp.toString()), sspSet); } if (!broadcastStreams.isEmpty()) { for (Set<SystemStreamPartition> value : groupedMap.values()) { value.addAll(broadcastStreams); } } return groupedMap; } GroupBySystemStreamPartition(Config config); @Override Map<TaskName, Set<SystemStreamPartition>> group(Set<SystemStreamPartition> ssps);  }
@Test public void testLocalStreamGroupedCorrectly() { SystemStreamPartitionGrouper grouper = grouperFactory.getSystemStreamPartitionGrouper(new MapConfig()); Map<TaskName, Set<SystemStreamPartition>> emptyResult = grouper.group(new HashSet<>()); assertTrue(emptyResult.isEmpty()); Map<TaskName, Set<SystemStreamPartition>> result = grouper.group(ImmutableSet.of(aa0, aa1, aa2, ac0)); Map<TaskName, Set<SystemStreamPartition>> expectedResult = ImmutableMap.<TaskName, Set<SystemStreamPartition>>builder() .put(new TaskName(aa0.toString()), ImmutableSet.of(aa0)) .put(new TaskName(aa1.toString()), ImmutableSet.of(aa1)) .put(new TaskName(aa2.toString()), ImmutableSet.of(aa2)) .put(new TaskName(ac0.toString()), ImmutableSet.of(ac0)) .build(); assertEquals(expectedResult, result); }
@Test public void testBroadcastStreamGroupedCorrectly() { Config config = new MapConfig(ImmutableMap.of("task.broadcast.inputs", "SystemA.StreamA#0")); SystemStreamPartitionGrouper grouper = new GroupBySystemStreamPartition(config); Map<TaskName, Set<SystemStreamPartition>> result = grouper.group(ImmutableSet.of(aa0, aa1, aa2, ac0)); Map<TaskName, Set<SystemStreamPartition>> expectedResult = ImmutableMap.<TaskName, Set<SystemStreamPartition>>builder() .put(new TaskName(aa1.toString()), ImmutableSet.of(aa1, aa0)) .put(new TaskName(aa2.toString()), ImmutableSet.of(aa2, aa0)) .put(new TaskName(ac0.toString()), ImmutableSet.of(ac0, aa0)) .build(); assertEquals(expectedResult, result); }