flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorContractTest.java
/**
	 * Verify that we neither invoke the trigger nor the window function if a timer
	 * for a non-existent window fires.
	 */
Verify that we neither invoke the trigger nor the window function if a timer
for a non-existent window fires.
Verify that we neither invoke the trigger nor the window function if a timer for a non - existent window fires .
private void testNoTimerFiringForPurgedWindow(final TimeDomainAdaptor timeAdaptor) throws Exception {
		WindowAssigner<Integer, TimeWindow> mockAssigner = mockTimeWindowAssigner();
		timeAdaptor.setIsEventTime(mockAssigner);
		Trigger<Integer, TimeWindow> mockTrigger = mockTrigger();
		@SuppressWarnings("unchecked")
		InternalWindowFunction<Iterable<Integer>, List<Integer>, Integer, TimeWindow> mockWindowFunction =
				mock(InternalWindowFunction.class);
		KeyedOneInputStreamOperatorTestHarness<Integer, Integer, List<Integer>> testHarness =
				createWindowOperator(mockAssigner, mockTrigger, 0L, mockWindowFunction);
		testHarness.open();
		timeAdaptor.advanceTime(testHarness, Long.MIN_VALUE);
		when(mockAssigner.assignWindows(anyInt(), anyLong(), anyAssignerContext()))
				.thenReturn(Arrays.asList(new TimeWindow(2, 4)));
		assertEquals(0, testHarness.extractOutputStreamRecords().size());
		assertEquals(0, testHarness.numKeyedStateEntries());
		doAnswer(new Answer<TriggerResult>() {
			@Override
			public TriggerResult answer(InvocationOnMock invocation) throws Exception {
				Trigger.TriggerContext context = (Trigger.TriggerContext) invocation.getArguments()[3];
				timeAdaptor.registerTimer(context, 0L);
				return TriggerResult.PURGE;
			}
		}).when(mockTrigger).onElement(Matchers.<Integer>anyObject(), anyLong(), anyTimeWindow(), anyTriggerContext());
		testHarness.processElement(new StreamRecord<>(0, 0L));
		assertEquals(0, testHarness.numKeyedStateEntries()); 
		assertEquals(2, timeAdaptor.numTimers(testHarness)); 
		timeAdaptor.advanceTime(testHarness, 0L);
		timeAdaptor.verifyTriggerCallback(mockTrigger, never(), null, null);
		verify(mockWindowFunction, never())
				.process(anyInt(), anyTimeWindow(), anyInternalWindowContext(), anyIntIterable(), WindowOperatorContractTest.<List<Integer>>anyCollector());
		assertEquals(1, timeAdaptor.numTimers(testHarness)); 
	}
private void testNoTimerFiringForPurgedWindow ( final TimeDomainAdaptor timeAdaptor ) throws Exception { WindowAssigner < Integer , TimeWindow > mockAssigner = mockTimeWindowAssigner ( ) ; timeAdaptor . setIsEventTime ( mockAssigner ) ; Trigger < Integer , TimeWindow > mockTrigger = mockTrigger ( ) ; @ SuppressWarnings ( "unchecked" ) InternalWindowFunction < Iterable < Integer > , List < Integer > , Integer , TimeWindow > mockWindowFunction = mock ( InternalWindowFunction . class ) ; KeyedOneInputStreamOperatorTestHarness < Integer , Integer , List < Integer > > testHarness = createWindowOperator ( mockAssigner , mockTrigger , 0L , mockWindowFunction ) ; testHarness . open ( ) ; timeAdaptor . advanceTime ( testHarness , Long . MIN_VALUE ) ; when ( mockAssigner . assignWindows ( anyInt ( ) , anyLong ( ) , anyAssignerContext ( ) ) ) . thenReturn ( Arrays . asList ( new TimeWindow ( 2 , 4 ) ) ) ; assertEquals ( 0 , testHarness . extractOutputStreamRecords ( ) . size ( ) ) ; assertEquals ( 0 , testHarness . numKeyedStateEntries ( ) ) ; doAnswer ( new Answer < TriggerResult > ( ) { @ Override public TriggerResult answer ( InvocationOnMock invocation ) throws Exception { Trigger . TriggerContext context = ( Trigger . TriggerContext ) invocation . getArguments ( ) [ 3 ] ; timeAdaptor . registerTimer ( context , 0L ) ; return TriggerResult . PURGE ; } } ) . when ( mockTrigger ) . onElement ( Matchers . < Integer > anyObject ( ) , anyLong ( ) , anyTimeWindow ( ) , anyTriggerContext ( ) ) ; testHarness . processElement ( new StreamRecord < > ( 0 , 0L ) ) ; assertEquals ( 0 , testHarness . numKeyedStateEntries ( ) ) ; assertEquals ( 2 , timeAdaptor . numTimers ( testHarness ) ) ; timeAdaptor . advanceTime ( testHarness , 0L ) ; timeAdaptor . verifyTriggerCallback ( mockTrigger , never ( ) , null , null ) ; verify ( mockWindowFunction , never ( ) ) . process ( anyInt ( ) , anyTimeWindow ( ) , anyInternalWindowContext ( ) , anyIntIterable ( ) , WindowOperatorContractTest . < List < Integer > > anyCollector ( ) ) ; assertEquals ( 1 , timeAdaptor . numTimers ( testHarness ) ) ; }
Verify that we neither invoke the trigger nor the window function if a timer
 for a non-existent window fires.
Verify that we neither invoke the trigger nor the window function if a timer for a non - existent window fires .
// don't interfere with cleanup timers // not contents or state // timer and gc timer // trigger is not called if there is no more window (timer is silently ignored) // only gc timers left