public Set<CoLocationGroup> getCoLocationGroups() {
        final Set<CoLocationGroup> coLocationGroups =
                IterableUtils.toStream(getVertices())
                        .map(JobVertex::getCoLocationGroup)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toSet());
        return Collections.unmodifiableSet(coLocationGroups);
    }
@Test
    public void testGetCoLocationGroups() {
        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 slotSharingGroup = new SlotSharingGroup();
        v1.setSlotSharingGroup(slotSharingGroup);
        v2.setSlotSharingGroup(slotSharingGroup);
        v1.setStrictlyCoLocatedWith(v2);

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

        assertThat(jobGraph.getCoLocationGroups(), hasSize(1));

        final CoLocationGroup onlyCoLocationGroup =
                jobGraph.getCoLocationGroups().iterator().next();
        assertThat(onlyCoLocationGroup.getVertexIds(), containsInAnyOrder(v1.getID(), v2.getID()));
    }