@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 testProcessingTimeTumblingWindows() throws Exception {
        final TumblingWindowAssigner assigner =
                TumblingWindowAssigner.of(Duration.ofSeconds(3)).withProcessingTime();
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
                createTestHarness(assigner, shiftTimeZone);
        testHarness.setup(OUT_SERIALIZER);
        testHarness.open();

        // process elements
        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
        // timestamp is ignored in processing time
        testHarness.setProcessingTime(20L);
        testHarness.processElement(insertRecord("key1", 1, Long.MAX_VALUE));
        testHarness.setProcessingTime(3999L);
        testHarness.processElement(insertRecord("key2", 1, Long.MAX_VALUE));
        // append 3 fields: window_start, window_end, window_time
        expectedOutput.add(
                insertRecord("key1", 1, Long.MAX_VALUE, localMills(0L), localMills(3000L), 2999L));
        expectedOutput.add(
                insertRecord(
                        "key2", 1, Long.MAX_VALUE, localMills(3000L), localMills(6000L), 5999L));
        ASSERTER.assertOutputEqualsSorted(
                "Output was not correct.", expectedOutput, testHarness.getOutput());
        testHarness.close();
    }