static JobVertexInputInfo computeVertexInputInfoForPointwise(
            int sourceCount,
            int targetCount,
            Function<Integer, Integer> numOfSubpartitionsRetriever,
            boolean isDynamicGraph) {

        final List<ExecutionVertexInputInfo> executionVertexInputInfos = new ArrayList<>();

        if (sourceCount >= targetCount) {
            for (int index = 0; index < targetCount; index++) {

                int start = index * sourceCount / targetCount;
                int end = (index + 1) * sourceCount / targetCount;

                IndexRange partitionRange = new IndexRange(start, end - 1);
                IndexRange subpartitionRange =
                        computeConsumedSubpartitionRange(
                                index,
                                1,
                                () -> numOfSubpartitionsRetriever.apply(start),
                                isDynamicGraph,
                                false);
                executionVertexInputInfos.add(
                        new ExecutionVertexInputInfo(index, partitionRange, subpartitionRange));
            }
        } else {
            for (int partitionNum = 0; partitionNum < sourceCount; partitionNum++) {

                int start = (partitionNum * targetCount + sourceCount - 1) / sourceCount;
                int end = ((partitionNum + 1) * targetCount + sourceCount - 1) / sourceCount;
                int numConsumers = end - start;

                IndexRange partitionRange = new IndexRange(partitionNum, partitionNum);
                // Variable used in lambda expression should be final or effectively final
                final int finalPartitionNum = partitionNum;
                for (int i = start; i < end; i++) {
                    IndexRange subpartitionRange =
                            computeConsumedSubpartitionRange(
                                    i,
                                    numConsumers,
                                    () -> numOfSubpartitionsRetriever.apply(finalPartitionNum),
                                    isDynamicGraph,
                                    false);
                    executionVertexInputInfos.add(
                            new ExecutionVertexInputInfo(i, partitionRange, subpartitionRange));
                }
            }
        }
        return new JobVertexInputInfo(executionVertexInputInfos);
    }
@Test
    void testComputeVertexInputInfoForPointwiseWithNonDynamicGraph() {
        final JobVertexInputInfo jobVertexInputInfo =
                computeVertexInputInfoForPointwise(2, 3, ignored -> 3, false);
        assertThat(jobVertexInputInfo.getExecutionVertexInputInfos())
                .containsExactlyInAnyOrder(
                        new ExecutionVertexInputInfo(0, new IndexRange(0, 0), new IndexRange(0, 0)),
                        new ExecutionVertexInputInfo(1, new IndexRange(0, 0), new IndexRange(1, 1)),
                        new ExecutionVertexInputInfo(
                                2, new IndexRange(1, 1), new IndexRange(0, 0)));
    }