@Override
    public void endInput(int inputId) throws Exception {
        inputSelectionHandler.endInput(inputId);
        InputSpec inputSpec = inputSpecMap.get(inputId);
        inputSpec.getOutput().endOperatorInput(inputSpec.getOutputOpInputId());
    }
@Test
    public void testProcess() throws Exception {
        TestingBatchMultipleInputStreamOperator op = createMultipleInputStreamOperator();
        List<StreamElement> outputData = op.getOutputData();

        TestingTwoInputStreamOperator joinOp2 =
                (TestingTwoInputStreamOperator) op.getTailWrapper().getStreamOperator();

        TableOperatorWrapper<?> joinWrapper1 = op.getTailWrapper().getInputWrappers().get(0);
        TestingTwoInputStreamOperator joinOp1 =
                (TestingTwoInputStreamOperator) joinWrapper1.getStreamOperator();

        TableOperatorWrapper<?> aggWrapper1 = joinWrapper1.getInputWrappers().get(0);
        TestingOneInputStreamOperator aggOp1 =
                (TestingOneInputStreamOperator) aggWrapper1.getStreamOperator();

        TableOperatorWrapper<?> aggWrapper2 = joinWrapper1.getInputWrappers().get(1);
        TestingOneInputStreamOperator aggOp2 =
                (TestingOneInputStreamOperator) aggWrapper2.getStreamOperator();

        List<Input> inputs = op.getInputs();
        assertThat(inputs).hasSize(3);
        Input input1 = inputs.get(0);
        Input input2 = inputs.get(1);
        Input input3 = inputs.get(2);

        assertThat(input1).isInstanceOf(OneInput.class);
        assertThat(input2).isInstanceOf(OneInput.class);
        assertThat(input3).isInstanceOf(SecondInputOfTwoInput.class);

        assertThat(joinOp2.getCurrentElement1()).isNull();
        assertThat(joinOp2.getCurrentElement2()).isNull();

        assertThat(joinOp1.getCurrentElement1()).isNull();
        assertThat(joinOp1.getCurrentElement2()).isNull();

        assertThat(aggOp1.getCurrentElement()).isNull();
        assertThat(aggOp2.getCurrentElement()).isNull();
        assertThat(outputData).isEmpty();

        // process first input (input id is 3)
        StreamRecord<RowData> element1 =
                new StreamRecord<>(GenericRowData.of(StringData.fromString("123")), 456);
        input3.processElement(element1);

        assertThat(joinOp2.getCurrentElement2()).isEqualTo(element1);
        assertThat(joinOp2.getCurrentElement1()).isNull();
        assertThat(outputData).isEmpty();

        // finish first input
        assertThat(joinOp2.getEndInputs()).isEmpty();
        op.endInput(3);
        assertThat(outputData).isEmpty();
        assertThat(joinOp2.getEndInputs()).containsExactly(2);

        // process second input (input id is 1)
        StreamRecord<RowData> element2 =
                new StreamRecord<>(GenericRowData.of(StringData.fromString("124")), 457);
        input1.processElement(element2);

        assertThat(aggOp1.getCurrentElement()).isEqualTo(element2);
        assertThat(joinOp1.getCurrentElement1()).isNull();
        assertThat(joinOp2.getCurrentElement1()).isNull();
        assertThat(outputData).isEmpty();

        // finish second input
        assertThat(joinOp1.getEndInputs()).isEmpty();
        op.endInput(1);
        assertThat(joinOp1.getEndInputs()).containsExactly(1);
        assertThat(joinOp2.getEndInputs()).containsExactly(2);
        assertThat(joinOp1.getCurrentElement1()).isEqualTo(element2);
        assertThat(outputData).isEmpty();

        // process third input (input id is 2)
        StreamRecord<RowData> element3 =
                new StreamRecord<>(GenericRowData.of(StringData.fromString("125")), 458);
        input2.processElement(element3);

        assertThat(aggOp2.getCurrentElement()).isEqualTo(element3);
        assertThat(joinOp1.getCurrentElement2()).isNull();
        assertThat(joinOp2.getCurrentElement1()).isNull();
        assertThat(outputData).isEmpty();

        // finish third input
        assertThat(joinOp1.getEndInputs()).containsExactly(1);
        op.endInput(2);
        assertThat(joinOp1.getEndInputs()).isEqualTo(Arrays.asList(1, 2));
        assertThat(joinOp2.getEndInputs()).isEqualTo(Arrays.asList(2, 1));
        assertThat(joinOp1.getCurrentElement2()).isEqualTo(element3);
        assertThat(outputData).hasSize(3);
    }