@Override
    public ProcessConfigurable<?> toSink(Sink<T> sink) {
        DataStreamV2SinkTransformation<T, T> sinkTransformation =
                StreamUtils.addSinkOperator(this, sink, getType());
        // Operator parallelism should always be 1 for global stream.
        // parallelismConfigured should be true to avoid overwritten by AdaptiveBatchScheduler.
        sinkTransformation.setParallelism(1, true);
        return StreamUtils.wrapWithConfigureHandle(
                new GlobalStreamImpl<>(environment, sinkTransformation));
    }
@Test
    void testToSink() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        GlobalStreamImpl<Integer> stream =
                new GlobalStreamImpl<>(env, new TestingTransformation<>("t1", Types.INT, 1));
        stream.toSink(DataStreamV2SinkUtils.wrapSink(new DiscardingSink<>()));
        List<Transformation<?>> transformations = env.getTransformations();
        assertThat(transformations)
                .hasSize(1)
                .element(0)
                .isInstanceOf(DataStreamV2SinkTransformation.class);
    }