public static <T, R> OneInputTransformation<T, R> getOneInputTransformation(
            String operatorName,
            AbstractDataStream<T> inputStream,
            TypeInformation<R> outTypeInformation,
            OneInputStreamOperator<T, R> operator) {
        // read the output type of the input Transform to coax out errors about MissingTypeInfo
        inputStream.getTransformation().getOutputType();

        OneInputTransformation<T, R> resultTransform =
                new OneInputTransformation<>(
                        inputStream.getTransformation(),
                        operatorName,
                        SimpleUdfStreamOperatorFactory.of(operator),
                        outTypeInformation,
                        inputStream.getEnvironment().getParallelism(),
                        false);

        return resultTransform;
    }
@Test
    void testGetOneInputTransformation() throws Exception {
        ExecutionEnvironmentImpl env = StreamTestUtils.getEnv();
        ProcessOperator<Integer, Long> operator =
                new ProcessOperator<>(new StreamTestUtils.NoOpOneInputStreamProcessFunction());
        OneInputTransformation<Integer, Long> transformation =
                StreamUtils.getOneInputTransformation(
                        "op",
                        new NonKeyedPartitionStreamImpl<>(
                                env, new TestingTransformation<>("t", Types.INT, 1)),
                        Types.LONG,
                        operator);
        assertThat(transformation.getOperator()).isEqualTo(operator);
        assertThat(transformation.getOutputType()).isEqualTo(Types.LONG);
        assertThat(transformation.getStateKeySelector()).isNull();
    }