test_class.file: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPeriodicWatermarksOperatorTest.java
test_class.identifier: TimestampsAndPeriodicWatermarksOperatorTest
test_case.body: @Test
  	public void testTimestampsAndPeriodicWatermarksOperator() throws Exception {
  		
  		final TimestampsAndPeriodicWatermarksOperator<Long> operator = 
  				new TimestampsAndPeriodicWatermarksOperator<Long>(new LongExtractor());
  
  		final ExecutionConfig config = new ExecutionConfig();
  		config.setAutoWatermarkInterval(50);
  		
  		OneInputStreamOperatorTestHarness<Long, Long> testHarness =
  				new OneInputStreamOperatorTestHarness<Long, Long>(operator, config);
  
  		testHarness.open();
  		
  		testHarness.processElement(new StreamRecord<>(1L, 1));
  		testHarness.processElement(new StreamRecord<>(2L, 1));
  		testHarness.processWatermark(new Watermark(2)); // this watermark should be ignored
  		testHarness.processElement(new StreamRecord<>(3L, 3));
  		testHarness.processElement(new StreamRecord<>(4L, 3));
  		
  		// 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();
  					assertNotNull(next);
  					Tuple2<Long, Long> update = validateElement(next, nextElementValue, lastWatermark);
  					nextElementValue = update.f0;
  					lastWatermark = update.f1;
  					
  					// check the invariant
  					assertTrue(lastWatermark < nextElementValue);
  				} else {
  					Thread.sleep(10);
  				}
  			}
  			
  			output.clear();
  		}
  
  		testHarness.processElement(new StreamRecord<>(4L, 4));
  		testHarness.processElement(new StreamRecord<>(5L, 4));
  		testHarness.processElement(new StreamRecord<>(6L, 4));
  		testHarness.processElement(new StreamRecord<>(7L, 4));
  		testHarness.processElement(new StreamRecord<>(8L, 4));
  
  		// 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();
  					assertNotNull(next);
  					Tuple2<Long, Long> update = validateElement(next, nextElementValue, lastWatermark);
  					nextElementValue = update.f0;
  					lastWatermark = update.f1;
  					
  					// check the invariant
  					assertTrue(lastWatermark < nextElementValue);
  				} else {
  					Thread.sleep(10);
  				}
  			}
  
  			output.clear();
  		}
  		
  		testHarness.processWatermark(new Watermark(Long.MAX_VALUE));
  		assertEquals(Long.MAX_VALUE, ((Watermark) testHarness.getOutput().poll()).getTimestamp());
  	}
test_case.class_method_signature: TimestampsAndPeriodicWatermarksOperatorTest.testTimestampsAndPeriodicWatermarksOperator()
test_case.full_signature: @Test public void testTimestampsAndPeriodicWatermarksOperator()
test_case.identifier: testTimestampsAndPeriodicWatermarksOperator
test_case.invocations: setAutoWatermarkInterval open processElement processElement processWatermark processElement processElement getOutput size poll assertNotNull validateElement assertTrue sleep clear processElement processElement processElement processElement processElement getOutput size poll assertNotNull validateElement assertTrue sleep clear processWatermark assertEquals getTimestamp poll getOutput
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testTimestampsAndPeriodicWatermarksOperator()
focal_class.declarator: serialVersionUID = 1L
focal_class.modifier: private static final
focal_class.original_string: private static final long serialVersionUID = 1L;
focal_class.type: long
focal_class.var_name: serialVersionUID
focal_class.declarator: watermarkInterval
focal_class.modifier: private transient
focal_class.original_string: private transient long watermarkInterval;
focal_class.type: long
focal_class.var_name: watermarkInterval
focal_class.declarator: currentWatermark
focal_class.modifier: private transient
focal_class.original_string: private transient long currentWatermark;
focal_class.type: long
focal_class.var_name: currentWatermark
focal_class.file: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPeriodicWatermarksOperator.java
focal_class.identifier: TimestampsAndPeriodicWatermarksOperator
focal_class.interfaces: implements OneInputStreamOperator<T, T>, Triggerable
focal_class.class_method_signature: TimestampsAndPeriodicWatermarksOperator.TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner)
focal_class.full_signature: public  TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner)
focal_class.identifier: TimestampsAndPeriodicWatermarksOperator
focal_class.modifiers: public
focal_class.parameters: (AssignerWithPeriodicWatermarks<T> assigner)
focal_class.signature:  TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner)
focal_class.class_method_signature: TimestampsAndPeriodicWatermarksOperator.open()
focal_class.full_signature: @Override public void open()
focal_class.identifier: open
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void open()
focal_class.class_method_signature: TimestampsAndPeriodicWatermarksOperator.processElement(StreamRecord<T> element)
focal_class.full_signature: @Override public void processElement(StreamRecord<T> element)
focal_class.identifier: processElement
focal_class.modifiers: @Override public
focal_class.parameters: (StreamRecord<T> element)
focal_class.return: void
focal_class.signature: void processElement(StreamRecord<T> element)
focal_class.class_method_signature: TimestampsAndPeriodicWatermarksOperator.trigger(long timestamp)
focal_class.full_signature: @Override public void trigger(long timestamp)
focal_class.identifier: trigger
focal_class.modifiers: @Override public
focal_class.parameters: (long timestamp)
focal_class.return: void
focal_class.signature: void trigger(long timestamp)
focal_class.class_method_signature: TimestampsAndPeriodicWatermarksOperator.processWatermark(Watermark mark)
focal_class.full_signature: @Override public void processWatermark(Watermark mark)
focal_class.identifier: processWatermark
focal_class.modifiers: @Override public
focal_class.parameters: (Watermark mark)
focal_class.return: void
focal_class.signature: void processWatermark(Watermark mark)
focal_class.class_method_signature: TimestampsAndPeriodicWatermarksOperator.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.superclass: extends AbstractUdfStreamOperator<T, AssignerWithPeriodicWatermarks<T>>
focal_method.body: public TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner) {
  		super(assigner);
  		this.chainingStrategy = ChainingStrategy.ALWAYS;
  	}
focal_method.class_method_signature: TimestampsAndPeriodicWatermarksOperator.TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner)
focal_method.full_signature: public  TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner)
focal_method.identifier: TimestampsAndPeriodicWatermarksOperator
focal_method.modifiers: public
focal_method.parameters: (AssignerWithPeriodicWatermarks<T> assigner)
focal_method.signature:  TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner)