@VisibleForTesting
    static VertexParallelismStore computeVertexParallelismStoreForExecution(
            JobGraph jobGraph,
            SchedulerExecutionMode executionMode,
            Function<JobVertex, Integer> defaultMaxParallelismFunc) {
        if (executionMode == SchedulerExecutionMode.REACTIVE) {
            return computeReactiveModeVertexParallelismStore(
                    jobGraph.getVertices(), defaultMaxParallelismFunc, false);
        }
        return SchedulerBase.computeVertexParallelismStore(
                jobGraph.getVertices(), defaultMaxParallelismFunc);
    }
@Test
    void testComputeVertexParallelismStoreForExecutionInReactiveMode() {
        JobVertex v1 = createNoOpVertex("v1", 1, 50);
        JobVertex v2 = createNoOpVertex("v2", 50, 50);
        JobGraph graph = streamingJobGraph(v1, v2);

        VertexParallelismStore parallelismStore =
                AdaptiveScheduler.computeVertexParallelismStoreForExecution(
                        graph,
                        SchedulerExecutionMode.REACTIVE,
                        SchedulerBase::getDefaultMaxParallelism);

        for (JobVertex vertex : graph.getVertices()) {
            VertexParallelismInformation info = parallelismStore.getParallelismInfo(vertex.getID());

            assertThat(info.getParallelism()).isEqualTo(vertex.getParallelism());
            assertThat(info.getMaxParallelism()).isEqualTo(vertex.getMaxParallelism());
        }
    }