public static <T, R, K> OneInputTransformation<T, R> getOneInputKeyedTransformation(
            String operatorName,
            AbstractDataStream<T> inputStream,
            TypeInformation<R> outTypeInformation,
            OneInputStreamOperator<T, R> operator,
            KeySelector<T, K> keySelector,
            TypeInformation<K> keyType) {
        OneInputTransformation<T, R> resultTransform =
                getOneInputTransformation(operatorName, inputStream, outTypeInformation, operator);

        // inject the key selector and key type
        resultTransform.setStateKeySelector(keySelector);
        resultTransform.setStateKeyType(keyType);

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