@VisibleForTesting
    public static JobGraph createJobGraph(StreamGraph streamGraph) {
        return new StreamingJobGraphGenerator(
                        Thread.currentThread().getContextClassLoader(),
                        streamGraph,
                        null,
                        Runnable::run)
                .createJobGraph();
    }
@Test
    void testSinkWithAllInterfaces() {
        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 TestSinkWithAllInterfaces()).name("sink");

        final StreamGraph streamGraph = env.getStreamGraph();
        final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
        assertThat(jobGraph.getNumberOfVertices()).isEqualTo(6);
        for (JobVertex jobVertex : jobGraph.getVertices()) {
            assertThat(jobVertex.getName())
                    .containsAnyOf(
                            "source",
                            "pre-writer",
                            "Writer",
                            "pre-committer",
                            "post-committer",
                            "Committer");
        }
    }