@Override
    public void processElement(StreamRecord<RowData> element) throws Exception {
        element.getValue().setRowKind(targetRowKind);
        output.collect(element);
    }
@Test
    public void testSetRowKind() throws Exception {
        // test set to all row kind
        for (RowKind targetRowKind : RowKind.values()) {
            RowKindSetter rowKindSetter = new RowKindSetter(targetRowKind);
            try (OneInputStreamOperatorTestHarness<RowData, RowData> operatorTestHarness =
                    new OneInputStreamOperatorTestHarness<>(rowKindSetter)) {
                operatorTestHarness.open();

                // get the rows with all row kind
                List<RowData> rows = getRowsWithAllRowKind();
                for (RowData row : rows) {
                    operatorTestHarness.processElement(new StreamRecord<>(row));
                }
                // verify the row kind of output
                verifyRowKind(operatorTestHarness.extractOutputValues(), targetRowKind);
            }
        }
    }