public CompletableFuture<Integer> inferSourceParallelismAsync(
            int parallelismInferenceUpperBound, long dataVolumePerTask) {
        return context.supplyAsync(
                        () -> {
                            if (!(source instanceof DynamicParallelismInference)) {
                                return ExecutionConfig.PARALLELISM_DEFAULT;
                            }

                            DynamicParallelismInference parallelismInference =
                                    (DynamicParallelismInference) source;
                            try {
                                return parallelismInference.inferParallelism(
                                        new DynamicParallelismInference.Context() {
                                            @Override
                                            public Optional<DynamicFilteringInfo>
                                                    getDynamicFilteringInfo() {
                                                return getSourceDynamicFilteringInfo();
                                            }

                                            @Override
                                            public int getParallelismInferenceUpperBound() {
                                                return parallelismInferenceUpperBound;
                                            }

                                            @Override
                                            public long getDataVolumePerTask() {
                                                return dataVolumePerTask;
                                            }
                                        });
                            } catch (Throwable e) {
                                LOG.error(
                                        "Unexpected error occurred when dynamically inferring source parallelism.",
                                        e);
                                return ExecutionConfig.PARALLELISM_DEFAULT;
                            }
                        })
                .thenApply(future -> (Integer) future);
    }
@Test
    public void testInferSourceParallelismAsync() throws Exception {
        final String listeningID = "testListeningID";

        class TestDynamicFilteringEvent implements SourceEvent, DynamicFilteringInfo {}

        CoordinatorStore store = new CoordinatorStoreImpl();
        store.putIfAbsent(listeningID, new SourceEventWrapper(new TestDynamicFilteringEvent()));
        final SourceCoordinator<?, ?> coordinator =
                new SourceCoordinator<>(
                        OPERATOR_NAME,
                        createMockSource(),
                        context,
                        store,
                        WatermarkAlignmentParams.WATERMARK_ALIGNMENT_DISABLED,
                        listeningID);
        assertThat(coordinator.inferSourceParallelismAsync(2, 1).get()).isEqualTo(2);
    }