@Override
    public ParallelismAndInputInfos decideParallelismAndInputInfosForVertex(
            JobVertexID jobVertexId,
            List<BlockingResultInfo> consumedResults,
            int vertexInitialParallelism,
            int vertexMinParallelism,
            int vertexMaxParallelism) {
        checkArgument(
                vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT
                        || vertexInitialParallelism > 0);
        checkArgument(
                vertexMinParallelism == ExecutionConfig.PARALLELISM_DEFAULT
                        || vertexMinParallelism > 0);
        checkArgument(
                vertexMaxParallelism > 0
                        && vertexMaxParallelism >= vertexInitialParallelism
                        && vertexMaxParallelism >= vertexMinParallelism);

        if (consumedResults.isEmpty()) {
            // source job vertex
            int parallelism =
                    vertexInitialParallelism > 0
                            ? vertexInitialParallelism
                            : computeSourceParallelismUpperBound(jobVertexId, vertexMaxParallelism);
            return new ParallelismAndInputInfos(parallelism, Collections.emptyMap());
        } else {
            int minParallelism = Math.max(globalMinParallelism, vertexMinParallelism);
            int maxParallelism = globalMaxParallelism;

            if (vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT
                    && vertexMaxParallelism < minParallelism) {
                LOG.info(
                        "The vertex maximum parallelism {} is smaller than the minimum parallelism {}. "
                                + "Use {} as the lower bound to decide parallelism of job vertex {}.",
                        vertexMaxParallelism,
                        minParallelism,
                        vertexMaxParallelism,
                        jobVertexId);
                minParallelism = vertexMaxParallelism;
            }
            if (vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT
                    && vertexMaxParallelism < maxParallelism) {
                LOG.info(
                        "The vertex maximum parallelism {} is smaller than the global maximum parallelism {}. "
                                + "Use {} as the upper bound to decide parallelism of job vertex {}.",
                        vertexMaxParallelism,
                        maxParallelism,
                        vertexMaxParallelism,
                        jobVertexId);
                maxParallelism = vertexMaxParallelism;
            }
            checkState(maxParallelism >= minParallelism);

            if (vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT
                    && areAllInputsAllToAll(consumedResults)
                    && !areAllInputsBroadcast(consumedResults)) {
                return decideParallelismAndEvenlyDistributeData(
                        jobVertexId,
                        consumedResults,
                        vertexInitialParallelism,
                        minParallelism,
                        maxParallelism);
            } else {
                return decideParallelismAndEvenlyDistributeSubpartitions(
                        jobVertexId,
                        consumedResults,
                        vertexInitialParallelism,
                        minParallelism,
                        maxParallelism);
            }
        }
    }
@Test
    void testParallelismAlreadyDecided() {
        final DefaultVertexParallelismAndInputInfosDecider decider =
                createDecider(MIN_PARALLELISM, MAX_PARALLELISM, DATA_VOLUME_PER_TASK);

        AllToAllBlockingResultInfo allToAllBlockingResultInfo =
                createAllToAllBlockingResultInfo(
                        new long[] {10L, 15L, 13L, 12L, 1L, 10L, 8L, 20L, 12L, 17L});
        ParallelismAndInputInfos parallelismAndInputInfos =
                decider.decideParallelismAndInputInfosForVertex(
                        new JobVertexID(),
                        Collections.singletonList(allToAllBlockingResultInfo),
                        3,
                        MIN_PARALLELISM,
                        MAX_PARALLELISM);

        assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(3);
        assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(1);

        checkAllToAllJobVertexInputInfo(
                Iterables.getOnlyElement(
                        parallelismAndInputInfos.getJobVertexInputInfos().values()),
                Arrays.asList(new IndexRange(0, 2), new IndexRange(3, 5), new IndexRange(6, 9)));
    }