@Override
    public void processElement(StreamRecord<RowData> element) throws Exception {
        RowData inputRow = element.getValue();
        long timestamp;
        if (windowAssigner.isEventTime()) {
            if (inputRow.isNullAt(rowtimeIndex)) {
                // null timestamp would be dropped
                numNullRowTimeRecordsDropped.inc();
                return;
            }
            timestamp = inputRow.getTimestamp(rowtimeIndex, 3).getMillisecond();
        } else {
            timestamp = getProcessingTimeService().getCurrentProcessingTime();
        }
        timestamp = toUtcTimestampMills(timestamp, shiftTimeZone);
        Collection<TimeWindow> elementWindows = windowAssigner.assignWindows(inputRow, timestamp);
        collect(inputRow, elementWindows);
    }
@Test
    public void testConsumingChangelogRecords() throws Exception {
        final TumblingWindowAssigner assigner = TumblingWindowAssigner.of(Duration.ofSeconds(3));
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
                createTestHarness(assigner, shiftTimeZone);
        testHarness.setup(OUT_SERIALIZER);
        testHarness.open();

        // process elements
        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
        testHarness.processElement(binaryRecord(RowKind.INSERT, "key1", 1, 20L));
        testHarness.processElement(binaryRecord(RowKind.UPDATE_BEFORE, "key1", 1, 30L));
        testHarness.processElement(binaryRecord(RowKind.UPDATE_AFTER, "key1", 1, 40L));
        testHarness.processWatermark(new Watermark(999));
        // append 3 fields: window_start, window_end, window_time
        expectedOutput.add(
                binaryRecord(
                        RowKind.INSERT, "key1", 1, 20L, localMills(0L), localMills(3000L), 2999L));
        expectedOutput.add(
                binaryRecord(
                        RowKind.UPDATE_BEFORE,
                        "key1",
                        1,
                        30L,
                        localMills(0L),
                        localMills(3000L),
                        2999L));
        expectedOutput.add(
                binaryRecord(
                        RowKind.UPDATE_AFTER,
                        "key1",
                        1,
                        40L,
                        localMills(0L),
                        localMills(3000L),
                        2999L));
        expectedOutput.add(new Watermark(999));

        ASSERTER.assertOutputEqualsSorted(
                "Output was not correct.", expectedOutput, testHarness.getOutput());

        testHarness.processWatermark(new Watermark(9999));
        expectedOutput.add(new Watermark(9999));

        ASSERTER.assertOutputEqualsSorted(
                "Output was not correct.", expectedOutput, testHarness.getOutput());

        // late records would not be dropped
        testHarness.processElement(binaryRecord(RowKind.DELETE, "key1", 1, 200L));

        expectedOutput.add(
                binaryRecord(
                        RowKind.DELETE, "key1", 1, 200L, localMills(0L), localMills(3000L), 2999L));
        ASSERTER.assertOutputEqualsSorted(
                "Output was not correct.", expectedOutput, testHarness.getOutput());
        testHarness.close();
    }