@Override
    public int computeSourceParallelismUpperBound(JobVertexID jobVertexId, int maxParallelism) {
        if (globalDefaultSourceParallelism > maxParallelism) {
            LOG.info(
                    "The global default source parallelism {} is larger than the maximum parallelism {}. "
                            + "Use {} as the upper bound parallelism of source job vertex {}.",
                    globalDefaultSourceParallelism,
                    maxParallelism,
                    maxParallelism,
                    jobVertexId);
            return maxParallelism;
        } else {
            return globalDefaultSourceParallelism;
        }
    }
@Test
    void testComputeSourceParallelismUpperBound() {
        Configuration configuration = new Configuration();
        configuration.setInteger(
                BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_DEFAULT_SOURCE_PARALLELISM,
                DEFAULT_SOURCE_PARALLELISM);
        VertexParallelismAndInputInfosDecider vertexParallelismAndInputInfosDecider =
                DefaultVertexParallelismAndInputInfosDecider.from(MAX_PARALLELISM, configuration);
        assertThat(
                        vertexParallelismAndInputInfosDecider.computeSourceParallelismUpperBound(
                                new JobVertexID(), VERTEX_MAX_PARALLELISM))
                .isEqualTo(DEFAULT_SOURCE_PARALLELISM);
    }