Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3661
Thanks, @manuzhang I had another look at your changes. I would merge them
now but simplify the tests and `WindowOperator` a little, if that's alright
with you.
In `WindowOperator` I would change `WindowContext` to query the
`InternalTimerService` directly, as in:
```
@Override
public long currentProcessingTime() {
return internalTimerService.currentProcessingTime();
}
@Override
public long currentWatermark() {
return internalTimerService.currentWatermark();
}
```
I would introduce a specific test, like this:
```
@Test
public void testEventTimeQuerying() throws Exception {
testCurrentTimeQuerying(new EventTimeAdaptor());
}
@Test
public void testProcessingTimeQuerying() throws Exception {
testCurrentTimeQuerying(new ProcessingTimeAdaptor());
}
public void testCurrentTimeQuerying(final TimeDomainAdaptor
timeAdaptor) throws Exception {
WindowAssigner<Integer, TimeWindow> mockAssigner =
mockTimeWindowAssigner();
timeAdaptor.setIsEventTime(mockAssigner);
Trigger<Integer, TimeWindow> mockTrigger = mockTrigger();
InternalWindowFunction<Iterable<Integer>, Void, Integer,
TimeWindow> mockWindowFunction = mockWindowFunction();
final KeyedOneInputStreamOperatorTestHarness<Integer, Integer,
Void> testHarness =
createWindowOperator(mockAssigner, mockTrigger,
20L, mockWindowFunction);
testHarness.open();
shouldFireOnElement(mockTrigger);
when(mockAssigner.assignWindows(anyInt(), anyLong(),
anyAssignerContext()))
.thenReturn(Arrays.asList(new TimeWindow(0,
20)));
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocationOnMock)
throws Throwable {
InternalWindowFunction.InternalWindowContext
context =
(InternalWindowFunction.InternalWindowContext)invocationOnMock.getArguments()[2];
timeAdaptor.verifyCorrectTime(testHarness,
context);
return null;
}
}).when(mockWindowFunction).process(anyInt(), anyTimeWindow(),
anyInternalWindowContext(), anyIntIterable(),
WindowOperatorContractTest.<Void>anyCollector());
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocationOnMock)
throws Throwable {
InternalWindowFunction.InternalWindowContext
context =
(InternalWindowFunction.InternalWindowContext)invocationOnMock.getArguments()[1];
timeAdaptor.verifyCorrectTime(testHarness,
context);
return null;
}
}).when(mockWindowFunction).clear(anyTimeWindow(),
anyInternalWindowContext());
timeAdaptor.advanceTime(testHarness, 10);
testHarness.processElement(new StreamRecord<>(0, 0L));
verify(mockWindowFunction, times(1)).process(anyInt(),
anyTimeWindow(), anyInternalWindowContext(), anyIntIterable(),
WindowOperatorContractTest.<Void>anyCollector());
timeAdaptor.advanceTime(testHarness, 100);
verify(mockWindowFunction, times(1)).clear(anyTimeWindow(),
anyInternalWindowContext());
}
```
What do you think? I would change your commit and commit as one thing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
