@VisibleForTesting
    static CompletableFuture<Integer> mergeDynamicParallelismFutures(
            List<CompletableFuture<Integer>> sourceParallelismFutures) {
        return sourceParallelismFutures.stream()
                .reduce(
                        CompletableFuture.completedFuture(ExecutionConfig.PARALLELISM_DEFAULT),
                        (a, b) -> a.thenCombine(b, Math::max));
    }
@Test
    void testMergeDynamicParallelismFutures() {
        List<CompletableFuture<Integer>> sourceParallelismFutures = new ArrayList<>();

        CompletableFuture<Integer> sourceParallelismFuture =
                CompletableFuture.completedFuture(ExecutionConfig.PARALLELISM_DEFAULT);
        sourceParallelismFutures.add(sourceParallelismFuture);

        // Testing scenarios without effective parallelism
        CompletableFuture<Integer> mergedSourceParallelismFuture =
                AdaptiveBatchScheduler.mergeDynamicParallelismFutures(sourceParallelismFutures);
        assertThat(mergedSourceParallelismFuture.join())
                .isEqualTo(ExecutionConfig.PARALLELISM_DEFAULT);

        CompletableFuture<Integer> sourceParallelismFuture1 = CompletableFuture.completedFuture(1);
        CompletableFuture<Integer> sourceParallelismFuture2 = CompletableFuture.completedFuture(2);
        CompletableFuture<Integer> sourceParallelismFuture4 = CompletableFuture.completedFuture(4);

        sourceParallelismFutures.add(sourceParallelismFuture1);
        sourceParallelismFutures.add(sourceParallelismFuture2);
        sourceParallelismFutures.add(sourceParallelismFuture4);

        // Testing scenarios with multiple sources in ExecutionJobVertex
        mergedSourceParallelismFuture =
                AdaptiveBatchScheduler.mergeDynamicParallelismFutures(sourceParallelismFutures);
        assertThat(mergedSourceParallelismFuture.join()).isEqualTo(4);
    }