public WatermarkAssignerOperator(
            int rowtimeFieldIndex,
            WatermarkGenerator watermarkGenerator,
            long idleTimeout,
            ProcessingTimeService processingTimeService) {

        this.rowtimeFieldIndex = rowtimeFieldIndex;
        this.watermarkGenerator = watermarkGenerator;

        this.idleTimeout = idleTimeout;
        this.chainingStrategy = ChainingStrategy.ALWAYS;

        this.processingTimeService = checkNotNull(processingTimeService);
    }
@Test
    public void testWatermarkAssignerOperator() throws Exception {
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
                createTestHarness(0, WATERMARK_GENERATOR, -1);

        testHarness.getExecutionConfig().setAutoWatermarkInterval(50);

        long currentTime = 0;

        testHarness.open();

        testHarness.processElement(new StreamRecord<>(GenericRowData.of(1L)));
        testHarness.processElement(new StreamRecord<>(GenericRowData.of(2L)));
        testHarness.processWatermark(new Watermark(2)); // this watermark should be ignored
        testHarness.processElement(new StreamRecord<>(GenericRowData.of(3L)));
        testHarness.processElement(new StreamRecord<>(GenericRowData.of(4L)));

        // validate first part of the sequence. we poll elements until our
        // watermark updates to "3", which must be the result of the "4" element.
        {
            ConcurrentLinkedQueue<Object> output = testHarness.getOutput();
            long nextElementValue = 1L;
            long lastWatermark = -1L;

            while (lastWatermark < 3) {
                if (output.size() > 0) {
                    Object next = output.poll();
                    assertThat(next).isNotNull();
                    Tuple2<Long, Long> update =
                            validateElement(next, nextElementValue, lastWatermark);
                    nextElementValue = update.f0;
                    lastWatermark = update.f1;

                    // check the invariant
                    assertThat(lastWatermark).isLessThan(nextElementValue);
                } else {
                    currentTime = currentTime + 10;
                    testHarness.setProcessingTime(currentTime);
                }
            }

            output.clear();
        }

        testHarness.processElement(new StreamRecord<>(GenericRowData.of(4L)));
        testHarness.processElement(new StreamRecord<>(GenericRowData.of(5L)));
        testHarness.processElement(new StreamRecord<>(GenericRowData.of(6L)));
        testHarness.processElement(new StreamRecord<>(GenericRowData.of(7L)));
        testHarness.processElement(new StreamRecord<>(GenericRowData.of(8L)));

        // validate the next part of the sequence. we poll elements until our
        // watermark updates to "7", which must be the result of the "8" element.
        {
            ConcurrentLinkedQueue<Object> output = testHarness.getOutput();
            long nextElementValue = 4L;
            long lastWatermark = 2L;

            while (lastWatermark < 7) {
                if (output.size() > 0) {
                    Object next = output.poll();
                    assertThat(next).isNotNull();
                    Tuple2<Long, Long> update =
                            validateElement(next, nextElementValue, lastWatermark);
                    nextElementValue = update.f0;
                    lastWatermark = update.f1;

                    // check the invariant
                    assertThat(lastWatermark).isLessThan(nextElementValue);
                } else {
                    currentTime = currentTime + 10;
                    testHarness.setProcessingTime(currentTime);
                }
            }

            output.clear();
        }

        testHarness.processWatermark(new Watermark(Long.MAX_VALUE));
        assertThat(((Watermark) testHarness.getOutput().poll()).getTimestamp())
                .isEqualTo(Long.MAX_VALUE);
    }