public void createOperator(StreamOperatorParameters<RowData> parameters) {
        checkArgument(wrapped == null, "This operator has been initialized");
        if (factory instanceof ProcessingTimeServiceAware) {
            ((ProcessingTimeServiceAware) factory)
                    .setProcessingTimeService(parameters.getProcessingTimeService());
        }
        wrapped = factory.createStreamOperator(parameters);
    }
@Test
    public void testCreateOperator() throws Exception {
        TestingOneInputStreamOperator operator = new TestingOneInputStreamOperator();
        TableOperatorWrapper<TestingOneInputStreamOperator> wrapper =
                createOneInputOperatorWrapper(operator, "test");
        StreamOperatorParameters<RowData> parameters = createStreamOperatorParameters();
        wrapper.createOperator(parameters);

        assertThat(wrapper.getStreamOperator()).isEqualTo(operator);

        // create operator again, will throw exception
        assertThatThrownBy(() -> wrapper.createOperator(parameters))
                .hasMessageContaining("This operator has been initialized");
    }