@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 testTumblingWindows() 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(insertRecord("key1", 1, 20L));
        testHarness.processElement(insertRecord("key2", 1, 3999L));
        testHarness.processWatermark(new Watermark(999));
        // append 3 fields: window_start, window_end, window_time
        expectedOutput.add(insertRecord("key1", 1, 20L, localMills(0L), localMills(3000L), 2999L));
        expectedOutput.add(
                insertRecord("key2", 1, 3999L, localMills(3000L), localMills(6000L), 5999L));
        expectedOutput.add(new Watermark(999));

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

        // late element would not be dropped
        testHarness.processElement(insertRecord("key2", 1, 80L));
        // rowtime is null, should be dropped
        testHarness.processElement(insertRecord("key2", 1, ((Long) null)));
        expectedOutput.add(insertRecord("key2", 1, 80L, localMills(0L), localMills(3000L), 2999L));
        ASSERTER.assertOutputEqualsSorted(
                "Output was not correct.", expectedOutput, testHarness.getOutput());

        assertThat(
                        ((AlignedWindowTableFunctionOperator) testHarness.getOperator())
                                .getNumNullRowTimeRecordsDropped()
                                .getCount())
                .isEqualTo(1);
        testHarness.close();
    }