@VisibleForTesting
    public static JobGraph createJobGraph(StreamGraph streamGraph) {
        return new StreamingJobGraphGenerator(
                        Thread.currentThread().getContextClassLoader(),
                        streamGraph,
                        null,
                        Runnable::run)
                .createJobGraph();
    }
@Test
    void testSinkSupportConcurrentExecutionAttempts() {
        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        final DataStream<Integer> source = env.fromData(1, 2, 3).name("source");
        source.rebalance()
                .sinkTo(new TestSinkWithSupportsConcurrentExecutionAttempts())
                .name("sink");

        final StreamGraph streamGraph = env.getStreamGraph();
        final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
        assertThat(jobGraph.getNumberOfVertices()).isEqualTo(6);
        for (JobVertex jobVertex : jobGraph.getVertices()) {
            if (jobVertex.getName().contains("source")) {
                assertThat(jobVertex.isSupportsConcurrentExecutionAttempts()).isTrue();
            } else if (jobVertex.getName().contains("pre-writer")) {
                assertThat(jobVertex.isSupportsConcurrentExecutionAttempts()).isTrue();
            } else if (jobVertex.getName().contains("Writer")) {
                assertThat(jobVertex.isSupportsConcurrentExecutionAttempts()).isTrue();
            } else if (jobVertex.getName().contains("pre-committer")) {
                assertThat(jobVertex.isSupportsConcurrentExecutionAttempts()).isFalse();
            } else if (jobVertex.getName().contains("post-committer")) {
                assertThat(jobVertex.isSupportsConcurrentExecutionAttempts()).isFalse();
            } else if (jobVertex.getName().contains("Committer")) {
                assertThat(jobVertex.isSupportsConcurrentExecutionAttempts()).isFalse();
            } else {
                Assertions.fail("Unexpected job vertex " + jobVertex.getName());
            }
        }
    }