You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2017/04/27 13:09:30 UTC
[2/2] flink git commit: [FLINK-4953] Allow access to "time" in
ProcessWindowFunction.Context
[FLINK-4953] Allow access to "time" in ProcessWindowFunction.Context
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/db759c53
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/db759c53
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/db759c53
Branch: refs/heads/master
Commit: db759c5303bef8b412a183b4bcdf6d11abb7f327
Parents: a448fc5
Author: manuzhang <ow...@gmail.com>
Authored: Sun Apr 2 19:27:14 2017 +0800
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Thu Apr 27 15:09:08 2017 +0200
----------------------------------------------------------------------
.../FoldApplyProcessWindowFunction.java | 6 +-
.../InternalProcessApplyWindowContext.java | 16 +++--
.../windowing/ProcessWindowFunction.java | 8 ++-
.../ReduceApplyProcessWindowFunction.java | 6 +-
.../windowing/AccumulatingKeyedTimePanes.java | 10 +++
.../operators/windowing/WindowOperator.java | 10 +++
.../functions/InternalProcessWindowContext.java | 10 +++
.../functions/InternalWindowFunction.java | 4 ++
.../FoldApplyProcessWindowFunctionTest.java | 10 +++
.../functions/InternalWindowFunctionTest.java | 64 ++++++++++++++---
.../windowing/WindowOperatorContractTest.java | 75 ++++++++++++++++++++
.../util/OneInputStreamOperatorTestHarness.java | 9 ++-
.../scala/function/ProcessWindowFunction.scala | 12 +++-
.../ScalaProcessWindowFunctionWrapper.scala | 8 +++
14 files changed, 225 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/FoldApplyProcessWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/FoldApplyProcessWindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/FoldApplyProcessWindowFunction.java
index 281afc7..1b2c2e2 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/FoldApplyProcessWindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/FoldApplyProcessWindowFunction.java
@@ -100,16 +100,14 @@ public class FoldApplyProcessWindowFunction<K, W extends Window, T, ACC, R>
}
this.ctx.window = context.window();
- this.ctx.windowState = context.windowState();
- this.ctx.globalState = context.globalState();
+ this.ctx.context = context;
windowFunction.process(key, ctx, Collections.singletonList(result), out);
}
@Override
public void clear(final Context context) throws Exception{
this.ctx.window = context.window();
- this.ctx.windowState = context.windowState();
- this.ctx.globalState = context.globalState();
+ this.ctx.context = context;
windowFunction.clear(ctx);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
index 0cf0a8d..fc31b07 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
@@ -34,8 +34,7 @@ public class InternalProcessApplyWindowContext<IN, OUT, KEY, W extends Window>
extends ProcessWindowFunction<IN, OUT, KEY, W>.Context {
W window;
- KeyedStateStore windowState;
- KeyedStateStore globalState;
+ ProcessWindowFunction.Context context;
InternalProcessApplyWindowContext(ProcessWindowFunction<IN, OUT, KEY, W> function) {
function.super();
@@ -47,12 +46,21 @@ public class InternalProcessApplyWindowContext<IN, OUT, KEY, W extends Window>
}
@Override
+ public long currentProcessingTime() {
+ return context.currentProcessingTime();
+ }
+
+ @Override
+ public long currentWatermark() {
+ return context.currentWatermark();
+ }
+
public KeyedStateStore windowState() {
- return windowState;
+ return context.windowState();
}
@Override
public KeyedStateStore globalState() {
- return globalState;
+ return context.globalState();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.java
index 980a94a..2c80e9e 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.java
@@ -63,10 +63,16 @@ public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> impl
*/
public abstract class Context implements java.io.Serializable {
/**
- * @return The window that is being evaluated.
+ * Returns the window that is being evaluated.
*/
public abstract W window();
+ /** Returns the current processing time. */
+ public abstract long currentProcessingTime();
+
+ /** Returns the current event-time watermark. */
+ public abstract long currentWatermark();
+
/**
* State accessor for per-key and per-window state.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceApplyProcessWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceApplyProcessWindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceApplyProcessWindowFunction.java
index f5694c3..836726d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceApplyProcessWindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/ReduceApplyProcessWindowFunction.java
@@ -59,16 +59,14 @@ public class ReduceApplyProcessWindowFunction<K, W extends Window, T, R>
}
this.ctx.window = context.window();
- this.ctx.windowState = context.windowState();
- this.ctx.globalState = context.globalState();
+ this.ctx.context = context;
windowFunction.process(k, ctx, Collections.singletonList(curr), out);
}
@Override
public void clear(final Context context) throws Exception {
this.ctx.window = context.window();
- this.ctx.windowState = context.windowState();
- this.ctx.globalState = context.globalState();
+ this.ctx.context = context;
windowFunction.clear(ctx);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingKeyedTimePanes.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingKeyedTimePanes.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingKeyedTimePanes.java
index 55ba1e3..4511b6b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingKeyedTimePanes.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingKeyedTimePanes.java
@@ -194,6 +194,16 @@ public class AccumulatingKeyedTimePanes<Type, Key, Result> extends AbstractKeyed
}
@Override
+ public long currentProcessingTime() {
+ throw new UnsupportedOperationException("current processing time is not supported in this context");
+ }
+
+ @Override
+ public long currentWatermark() {
+ throw new UnsupportedOperationException("current watermark is not supported in this context");
+ }
+
+ @Override
public KeyedStateStore windowState() {
return throwingStore;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
index 8d8f6b5..a62c9ff 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
@@ -817,6 +817,16 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
}
@Override
+ public long currentProcessingTime() {
+ return internalTimerService.currentProcessingTime();
+ }
+
+ @Override
+ public long currentWatermark() {
+ return internalTimerService.currentWatermark();
+ }
+
+ @Override
public KeyedStateStore windowState() {
this.windowState.window = this.window;
return this.windowState;
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalProcessWindowContext.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalProcessWindowContext.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalProcessWindowContext.java
index 0f1c0ee..d575f01 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalProcessWindowContext.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalProcessWindowContext.java
@@ -47,6 +47,16 @@ public class InternalProcessWindowContext<IN, OUT, KEY, W extends Window>
}
@Override
+ public long currentProcessingTime() {
+ return internalContext.currentProcessingTime();
+ }
+
+ @Override
+ public long currentWatermark() {
+ return internalContext.currentWatermark();
+ }
+
+ @Override
public KeyedStateStore windowState() {
return internalContext.windowState();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalWindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalWindowFunction.java
index 4055c4f..ea64dda 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalWindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalWindowFunction.java
@@ -55,6 +55,10 @@ public interface InternalWindowFunction<IN, OUT, KEY, W extends Window> extends
* for internal use.
*/
interface InternalWindowContext extends java.io.Serializable {
+ long currentProcessingTime();
+
+ long currentWatermark();
+
KeyedStateStore windowState();
KeyedStateStore globalState();
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/FoldApplyProcessWindowFunctionTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/FoldApplyProcessWindowFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/FoldApplyProcessWindowFunctionTest.java
index 7cac1e1..90470ac 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/FoldApplyProcessWindowFunctionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/FoldApplyProcessWindowFunctionTest.java
@@ -154,6 +154,16 @@ public class FoldApplyProcessWindowFunctionTest {
}
@Override
+ public long currentProcessingTime() {
+ return 0;
+ }
+
+ @Override
+ public long currentWatermark() {
+ return 0;
+ }
+
+ @Override
public KeyedStateStore windowState() {
return new DummyKeyedStateStore();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java
index 6f34607..4b8057f 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/windowing/functions/InternalWindowFunctionTest.java
@@ -21,7 +21,6 @@ package org.apache.flink.streaming.api.operators.windowing.functions;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.common.state.KeyedStateStore;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
@@ -33,7 +32,6 @@ import org.apache.flink.streaming.api.functions.windowing.RichProcessWindowFunct
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.operators.OutputTypeConfigurable;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
-import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalAggregateProcessAllWindowFunction;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalAggregateProcessWindowFunction;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalIterableAllWindowFunction;
@@ -48,6 +46,8 @@ import org.apache.flink.streaming.runtime.operators.windowing.functions.Internal
import org.apache.flink.util.Collector;
import org.hamcrest.collection.IsIterableContainingInOrder;
import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
import java.util.HashMap;
import java.util.HashSet;
@@ -62,6 +62,7 @@ import static org.hamcrest.collection.IsMapContaining.hasEntry;
import static org.hamcrest.core.AllOf.allOf;
import static org.mockito.Mockito.anyObject;
import static org.mockito.Mockito.argThat;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -229,8 +230,24 @@ public class InternalWindowFunctionTest {
Collector<String> c = (Collector<String>) mock(Collector.class);
InternalWindowFunction.InternalWindowContext ctx = mock(InternalWindowFunction.InternalWindowContext.class);
+ doAnswer(new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+ ProcessWindowFunctionMock.Context c =
+ (ProcessWindowFunction.Context) invocationOnMock.getArguments()[1];
+ c.currentProcessingTime();
+ c.currentWatermark();
+ c.windowState();
+ c.globalState();
+ return null;
+ }
+ }).when(mock).process(eq(42L), (ProcessWindowFunctionMock.Context) anyObject(), eq(i), eq(c));
+
windowFunction.process(42L, w, ctx, i, c);
- verify(mock).process(eq(42L), (ProcessWindowFunctionMock.Context) anyObject(), eq(i), eq(c));
+ verify(ctx).currentProcessingTime();
+ verify(ctx).currentWatermark();
+ verify(ctx).windowState();
+ verify(ctx).globalState();
// check close
windowFunction.close();
@@ -396,8 +413,25 @@ public class InternalWindowFunctionTest {
Collector<String> c = (Collector<String>) mock(Collector.class);
InternalWindowFunction.InternalWindowContext ctx = mock(InternalWindowFunction.InternalWindowContext.class);
+ doAnswer(new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+ ProcessWindowFunctionMock.Context c =
+ (ProcessWindowFunction.Context) invocationOnMock.getArguments()[1];
+ c.currentProcessingTime();
+ c.currentWatermark();
+ c.windowState();
+ c.globalState();
+ return null;
+ }
+ }).when(mock).process(eq(42L), (ProcessWindowFunctionMock.Context) anyObject(),
+ (Iterable<Long>)argThat(IsIterableContainingInOrder.contains(23L)), eq(c));
+
windowFunction.process(42L, w, ctx,23L, c);
- verify(mock).process(eq(42L), (ProcessWindowFunctionMock.Context) anyObject(), (Iterable<Long>)argThat(IsIterableContainingInOrder.contains(23L)), eq(c));
+ verify(ctx).currentProcessingTime();
+ verify(ctx).currentWatermark();
+ verify(ctx).windowState();
+ verify(ctx).globalState();
// check close
windowFunction.close();
@@ -469,8 +503,18 @@ public class InternalWindowFunctionTest {
args.add(24L);
InternalWindowFunction.InternalWindowContext ctx = mock(InternalWindowFunction.InternalWindowContext.class);
- windowFunction.process(42L, w, ctx, args, c);
- verify(mock).process(
+ doAnswer(new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+ ProcessWindowFunctionMock.Context c =
+ (ProcessWindowFunction.Context) invocationOnMock.getArguments()[1];
+ c.currentProcessingTime();
+ c.currentWatermark();
+ c.windowState();
+ c.globalState();
+ return null;
+ }
+ }).when(mock).process(
eq(42L),
(AggregateProcessWindowFunctionMock.Context) anyObject(),
(Iterable) argThat(containsInAnyOrder(allOf(
@@ -478,6 +522,12 @@ public class InternalWindowFunctionTest {
hasEntry(is(24L), is(24L))))),
eq(c));
+ windowFunction.process(42L, w, ctx, args, c);
+ verify(ctx).currentProcessingTime();
+ verify(ctx).currentWatermark();
+ verify(ctx).windowState();
+ verify(ctx).globalState();
+
// check close
windowFunction.close();
verify(mock).close();
@@ -572,7 +622,6 @@ public class InternalWindowFunctionTest {
@Override
public void process(Long aLong, ProcessWindowFunction<Long, String, Long, TimeWindow>.Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
-
}
}
@@ -587,7 +636,6 @@ public class InternalWindowFunctionTest {
@Override
public void process(Long aLong, ProcessWindowFunction<Map<Long, Long>, String, Long, TimeWindow>.Context context, Iterable<Map<Long, Long>> elements, Collector<String> out) throws Exception {
-
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorContractTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorContractTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorContractTest.java
index 3ae8f37..0d80605 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorContractTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorContractTest.java
@@ -45,6 +45,8 @@ import java.util.List;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.MergingWindowAssigner;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
@@ -2450,6 +2452,61 @@ public abstract class WindowOperatorContractTest extends TestLogger {
testHarness.processElement(new StreamRecord<>(0, 0L));
}
+ @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());
+ }
+
protected abstract <W extends Window, OUT> KeyedOneInputStreamOperatorTestHarness<Integer, Integer, OUT> createWindowOperator(
WindowAssigner<Integer, W> assigner,
Trigger<Integer, W> trigger,
@@ -2495,6 +2552,10 @@ public abstract class WindowOperatorContractTest extends TestLogger {
VerificationMode verificationMode,
Long time,
TimeWindow window) throws Exception;
+
+ void verifyCorrectTime(
+ OneInputStreamOperatorTestHarness testHarness,
+ InternalWindowFunction.InternalWindowContext context);
}
private static class EventTimeAdaptor implements TimeDomainAdaptor {
@@ -2588,6 +2649,13 @@ public abstract class WindowOperatorContractTest extends TestLogger {
anyTriggerContext());
}
}
+
+ @Override
+ public void verifyCorrectTime(
+ OneInputStreamOperatorTestHarness testHarness,
+ InternalWindowFunction.InternalWindowContext context) {
+ assertEquals(testHarness.getCurrentWatermark(), context.currentWatermark());
+ }
}
private static class ProcessingTimeAdaptor implements TimeDomainAdaptor {
@@ -2681,5 +2749,12 @@ public abstract class WindowOperatorContractTest extends TestLogger {
anyTriggerContext());
}
}
+
+ @Override
+ public void verifyCorrectTime(
+ OneInputStreamOperatorTestHarness testHarness,
+ InternalWindowFunction.InternalWindowContext context) {
+ assertEquals(testHarness.getProcessingTime(), context.currentProcessingTime());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
index 9b02378..ced8cca 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
@@ -38,6 +38,8 @@ public class OneInputStreamOperatorTestHarness<IN, OUT>
private final OneInputStreamOperator<IN, OUT> oneInputOperator;
+ private long currentWatermark;
+
public OneInputStreamOperatorTestHarness(
OneInputStreamOperator<IN, OUT> operator,
TypeSerializer<IN> typeSerializerIn) throws Exception {
@@ -98,10 +100,15 @@ public class OneInputStreamOperatorTestHarness<IN, OUT>
}
public void processWatermark(long watermark) throws Exception {
- oneInputOperator.processWatermark(new Watermark(watermark));
+ processWatermark(new Watermark(watermark));
}
public void processWatermark(Watermark mark) throws Exception {
+ currentWatermark = mark.getTimestamp();
oneInputOperator.processWatermark(mark);
}
+
+ public long getCurrentWatermark() {
+ return currentWatermark;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/ProcessWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/ProcessWindowFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/ProcessWindowFunction.scala
index bdf6ae6..bc79a26 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/ProcessWindowFunction.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/ProcessWindowFunction.scala
@@ -63,11 +63,21 @@ abstract class ProcessWindowFunction[IN, OUT, KEY, W <: Window] extends Function
*/
abstract class Context {
/**
- * @return The window that is being evaluated.
+ * Returns the window that is being evaluated.
*/
def window: W
/**
+ * Returns the current processing time.
+ */
+ def currentProcessingTime: Long
+
+ /**
+ * Returns the current event-time watermark.
+ */
+ def currentWatermark: Long
+
+ /**
* State accessor for per-key and per-window state.
*/
def windowState: KeyedStateStore
http://git-wip-us.apache.org/repos/asf/flink/blob/db759c53/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaProcessWindowFunctionWrapper.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaProcessWindowFunctionWrapper.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaProcessWindowFunctionWrapper.scala
index fac5958..263373e 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaProcessWindowFunctionWrapper.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaProcessWindowFunctionWrapper.scala
@@ -53,6 +53,10 @@ final class ScalaProcessWindowFunctionWrapper[IN, OUT, KEY, W <: Window](
val ctx = new func.Context {
override def window = context.window
+ override def currentProcessingTime = context.currentProcessingTime
+
+ override def currentWatermark = context.currentWatermark
+
override def windowState = context.windowState()
override def globalState = context.globalState()
@@ -64,6 +68,10 @@ final class ScalaProcessWindowFunctionWrapper[IN, OUT, KEY, W <: Window](
val ctx = new func.Context {
override def window = context.window
+ override def currentProcessingTime = context.currentProcessingTime
+
+ override def currentWatermark = context.currentWatermark
+
override def windowState = context.windowState()
override def globalState = context.globalState()