static JobVertexInputInfo computeVertexInputInfoForAllToAll(
            int sourceCount,
            int targetCount,
            Function<Integer, Integer> numOfSubpartitionsRetriever,
            boolean isDynamicGraph,
            boolean isBroadcast) {
        final List<ExecutionVertexInputInfo> executionVertexInputInfos = new ArrayList<>();
        IndexRange partitionRange = new IndexRange(0, sourceCount - 1);
        for (int i = 0; i < targetCount; ++i) {
            IndexRange subpartitionRange =
                    computeConsumedSubpartitionRange(
                            i,
                            targetCount,
                            () -> numOfSubpartitionsRetriever.apply(0),
                            isDynamicGraph,
                            isBroadcast);
            executionVertexInputInfos.add(
                    new ExecutionVertexInputInfo(i, partitionRange, subpartitionRange));
        }
        return new JobVertexInputInfo(executionVertexInputInfos);
    }
@Test
    void testComputeVertexInputInfoForAllToAllWithNonDynamicGraph() {
        final JobVertexInputInfo nonBroadcast =
                computeVertexInputInfoForAllToAll(2, 3, ignored -> 3, false, false);
        assertThat(nonBroadcast.getExecutionVertexInputInfos())
                .containsExactlyInAnyOrder(
                        new ExecutionVertexInputInfo(0, new IndexRange(0, 1), new IndexRange(0, 0)),
                        new ExecutionVertexInputInfo(1, new IndexRange(0, 1), new IndexRange(1, 1)),
                        new ExecutionVertexInputInfo(
                                2, new IndexRange(0, 1), new IndexRange(2, 2)));

        final JobVertexInputInfo broadcast =
                computeVertexInputInfoForAllToAll(2, 3, ignored -> 3, false, true);
        assertThat(broadcast.getExecutionVertexInputInfos())
                .containsExactlyInAnyOrder(
                        new ExecutionVertexInputInfo(0, new IndexRange(0, 1), new IndexRange(0, 0)),
                        new ExecutionVertexInputInfo(1, new IndexRange(0, 1), new IndexRange(1, 1)),
                        new ExecutionVertexInputInfo(
                                2, new IndexRange(0, 1), new IndexRange(2, 2)));
    }