private static void connectAllToAll(
            ExecutionJobVertex jobVertex,
            IntermediateResult result,
            JobVertexInputInfo jobVertexInputInfo) {
        // check the vertex input info is legal
        jobVertexInputInfo
                .getExecutionVertexInputInfos()
                .forEach(
                        executionVertexInputInfo -> {
                            IndexRange partitionRange =
                                    executionVertexInputInfo.getPartitionIndexRange();
                            checkArgument(partitionRange.getStartIndex() == 0);
                            checkArgument(
                                    partitionRange.getEndIndex()
                                            == (result.getNumberOfAssignedPartitions() - 1));
                        });

        connectInternal(
                Arrays.asList(jobVertex.getTaskVertices()),
                Arrays.asList(result.getPartitions()),
                result.getResultType(),
                jobVertex.getGraph().getEdgeManager());
    }
@Test
    void testConnectAllToAll() throws Exception {
        int upstream = 3;
        int downstream = 2;

        // use dynamic graph to specify the vertex input info
        ExecutionGraph eg = setupExecutionGraph(upstream, downstream, POINTWISE, true);

        List<ExecutionVertexInputInfo> executionVertexInputInfos = new ArrayList<>();
        for (int i = 0; i < downstream; i++) {
            executionVertexInputInfos.add(
                    new ExecutionVertexInputInfo(
                            i,
                            new IndexRange(0, upstream - 1),
                            // the subpartition range will not be used in edge manager, so set (0,
                            // 0)
                            new IndexRange(0, 0)));
        }
        final JobVertexInputInfo jobVertexInputInfo =
                new JobVertexInputInfo(executionVertexInputInfos);

        final Iterator<ExecutionJobVertex> vertexIterator =
                eg.getVerticesTopologically().iterator();
        final ExecutionJobVertex producer = vertexIterator.next();
        final ExecutionJobVertex consumer = vertexIterator.next();

        // initialize producer and consumer
        eg.initializeJobVertex(producer, 1L, Collections.emptyMap());
        eg.initializeJobVertex(
                consumer,
                1L,
                Collections.singletonMap(
                        producer.getProducedDataSets()[0].getId(), jobVertexInputInfo));

        IntermediateResult result =
                Objects.requireNonNull(eg.getJobVertex(producer.getJobVertexId()))
                        .getProducedDataSets()[0];
        IntermediateResultPartition partition1 = result.getPartitions()[0];
        IntermediateResultPartition partition2 = result.getPartitions()[1];
        IntermediateResultPartition partition3 = result.getPartitions()[2];

        ExecutionVertex vertex1 = consumer.getTaskVertices()[0];
        ExecutionVertex vertex2 = consumer.getTaskVertices()[1];

        // check consumers of the partitions
        ConsumerVertexGroup consumerVertexGroup = partition1.getConsumerVertexGroups().get(0);
        assertThat(consumerVertexGroup).containsExactlyInAnyOrder(vertex1.getID(), vertex2.getID());
        assertThat(partition2.getConsumerVertexGroups().get(0)).isEqualTo(consumerVertexGroup);
        assertThat(partition3.getConsumerVertexGroups().get(0)).isEqualTo(consumerVertexGroup);

        // check inputs of the execution vertices
        ConsumedPartitionGroup consumedPartitionGroup = vertex1.getConsumedPartitionGroup(0);
        assertThat(consumedPartitionGroup)
                .containsExactlyInAnyOrder(
                        partition1.getPartitionId(),
                        partition2.getPartitionId(),
                        partition3.getPartitionId());
        assertThat(vertex2.getConsumedPartitionGroup(0)).isEqualTo(consumedPartitionGroup);

        // check the consumerVertexGroup and consumedPartitionGroup are set to each other
        assertThat(consumerVertexGroup.getConsumedPartitionGroup())
                .isEqualTo(consumedPartitionGroup);
        assertThat(consumedPartitionGroup.getConsumerVertexGroup()).isEqualTo(consumerVertexGroup);
    }