public Set<SlotSharingGroup> getSlotSharingGroups() {
        final Set<SlotSharingGroup> slotSharingGroups =
                IterableUtils.toStream(getVertices())
                        .map(JobVertex::getSlotSharingGroup)
                        .collect(Collectors.toSet());
        return Collections.unmodifiableSet(slotSharingGroups);
    }
@Test
    public void testGetSlotSharingGroups() {
        final JobVertex v1 = new JobVertex("1");
        final JobVertex v2 = new JobVertex("2");
        final JobVertex v3 = new JobVertex("3");
        final JobVertex v4 = new JobVertex("4");

        final SlotSharingGroup group1 = new SlotSharingGroup();
        v1.setSlotSharingGroup(group1);
        v2.setSlotSharingGroup(group1);

        final SlotSharingGroup group2 = new SlotSharingGroup();
        v3.setSlotSharingGroup(group2);
        v4.setSlotSharingGroup(group2);

        final JobGraph jobGraph =
                JobGraphBuilder.newStreamingJobGraphBuilder()
                        .addJobVertices(Arrays.asList(v1, v2, v3, v4))
                        .build();

        assertThat(jobGraph.getSlotSharingGroups(), containsInAnyOrder(group1, group2));
    }