flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [2/2] flink git commit: [hotfix] Add Window Parameter in Trigger.onEventTime/onProcessingTime
Date Fri, 23 Oct 2015 15:38:49 GMT
[hotfix] Add Window Parameter in Trigger.onEventTime/onProcessingTime

Before these trigger methods had no information about the window that
they are responsible for. This information might be required for
implementing more advanced trigger behaviour.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bc5b852a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bc5b852a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bc5b852a

Branch: refs/heads/release-0.10
Commit: bc5b852a29bd76a53975f40909e5259e034c9980
Parents: 15d3f10
Author: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Authored: Fri Oct 23 11:28:58 2015 +0200
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Fri Oct 23 17:38:24 2015 +0200

----------------------------------------------------------------------
 .../streaming/examples/windowing/SessionWindowing.java  |  5 ++---
 .../api/windowing/assigners/GlobalWindows.java          |  4 ++--
 .../windowing/triggers/ContinuousEventTimeTrigger.java  |  5 ++---
 .../triggers/ContinuousProcessingTimeTrigger.java       |  5 ++---
 .../streaming/api/windowing/triggers/CountTrigger.java  |  5 ++---
 .../streaming/api/windowing/triggers/DeltaTrigger.java  |  5 ++---
 .../api/windowing/triggers/EventTimeTrigger.java        | 12 +++++++-----
 .../api/windowing/triggers/ProcessingTimeTrigger.java   |  5 ++---
 .../api/windowing/triggers/PurgingTrigger.java          |  8 ++++----
 .../flink/streaming/api/windowing/triggers/Trigger.java | 12 ++++++------
 .../operators/windowing/NonKeyedWindowOperator.java     |  4 ++--
 .../runtime/operators/windowing/WindowOperator.java     |  4 ++--
 .../apache/flink/streaming/util/TestHarnessUtil.java    |  1 -
 13 files changed, 35 insertions(+), 40 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
----------------------------------------------------------------------
diff --git a/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
b/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
index 3c63156..035727a 100644
--- a/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
+++ b/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
@@ -126,7 +126,7 @@ public class SessionWindowing {
 		}
 
 		@Override
-		public TriggerResult onEventTime(long time, TriggerContext ctx) throws Exception {
+		public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws
Exception {
 			OperatorState<Long> lastSeenState = ctx.getKeyValueState("last-seen", 1L);
 			Long lastSeen = lastSeenState.value();
 
@@ -137,8 +137,7 @@ public class SessionWindowing {
 		}
 
 		@Override
-		public TriggerResult onProcessingTime(long time,
-				TriggerContext ctx) throws Exception {
+		public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx)
throws Exception {
 			return TriggerResult.CONTINUE;
 		}
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
index 4d5b9d7..99a4962 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java
@@ -79,12 +79,12 @@ public class GlobalWindows extends WindowAssigner<Object, GlobalWindow>
{
 		}
 
 		@Override
-		public TriggerResult onEventTime(long time, TriggerContext ctx) {
+		public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) {
 			return TriggerResult.CONTINUE;
 		}
 
 		@Override
-		public TriggerResult onProcessingTime(long time, TriggerContext ctx) {
+		public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx)
{
 			return TriggerResult.CONTINUE;
 		}
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
index ea26309..4b6af8f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
@@ -57,14 +57,13 @@ public class ContinuousEventTimeTrigger<W extends Window> implements
Trigger<Obj
 	}
 
 	@Override
-	public TriggerResult onEventTime(long time, TriggerContext ctx) {
+	public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
 		ctx.registerEventTimeTimer(time + interval);
 		return TriggerResult.FIRE;
 	}
 
 	@Override
-	public TriggerResult onProcessingTime(long time,
-			TriggerContext ctx) throws Exception {
+	public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception
{
 		return TriggerResult.CONTINUE;
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
index be56738..66f9bda 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java
@@ -63,13 +63,12 @@ public class ContinuousProcessingTimeTrigger<W extends Window> implements
Trigge
 	}
 
 	@Override
-	public TriggerResult onEventTime(long time,
-			TriggerContext ctx) throws Exception {
+	public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception
{
 		return TriggerResult.CONTINUE;
 	}
 
 	@Override
-	public TriggerResult onProcessingTime(long time, TriggerContext ctx) throws Exception {
+	public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception
{
 
 		OperatorState<Long> fireState = ctx.getKeyValueState("fire-timestamp", 0L);
 		long nextFireTimestamp = fireState.value();

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
index 8512989..efb62d7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java
@@ -49,13 +49,12 @@ public class CountTrigger<W extends Window> implements Trigger<Object,
W> {
 	}
 
 	@Override
-	public TriggerResult onEventTime(long time, TriggerContext ctx) {
+	public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
 		return TriggerResult.CONTINUE;
 	}
 
 	@Override
-	public TriggerResult onProcessingTime(long time,
-			TriggerContext ctx) throws Exception {
+	public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception
{
 		return TriggerResult.CONTINUE;
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
index 1c6523d..d791d28 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java
@@ -59,13 +59,12 @@ public class DeltaTrigger<T extends Serializable, W extends Window>
implements T
 	}
 
 	@Override
-	public TriggerResult onEventTime(long time, TriggerContext ctx) {
+	public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
 		return TriggerResult.CONTINUE;
 	}
 
 	@Override
-	public TriggerResult onProcessingTime(long time,
-			TriggerContext ctx) throws Exception {
+	public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception
{
 		return TriggerResult.CONTINUE;
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
index 4b6613c..831e360 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
@@ -37,13 +37,12 @@ public class EventTimeTrigger implements Trigger<Object, TimeWindow>
{
 	}
 
 	@Override
-	public TriggerResult onEventTime(long time, TriggerContext ctx) {
+	public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
 		return TriggerResult.FIRE_AND_PURGE;
 	}
 
 	@Override
-	public TriggerResult onProcessingTime(long time,
-			TriggerContext ctx) throws Exception {
+	public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx)
throws Exception {
 		return TriggerResult.CONTINUE;
 	}
 
@@ -53,10 +52,13 @@ public class EventTimeTrigger implements Trigger<Object, TimeWindow>
{
 	}
 
 	/**
-	 * Creates trigger that fires once the watermark passes the end of the window.
+	 * Creates an event-time trigger that fires once the watermark passes the end of the window.
+	 *
+	 * <p>
+	 * Once the trigger fires all elements are discarded. Elements that arrive late immediately
+	 * trigger window evaluation with just this one element.
 	 */
 	public static EventTimeTrigger create() {
 		return new EventTimeTrigger();
 	}
-
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
index 6278ba6..b460c8a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java
@@ -35,13 +35,12 @@ public class ProcessingTimeTrigger implements Trigger<Object, TimeWindow>
{
 	}
 
 	@Override
-	public TriggerResult onEventTime(long time,
-			TriggerContext ctx) throws Exception {
+	public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws
Exception {
 		return TriggerResult.CONTINUE;
 	}
 
 	@Override
-	public TriggerResult onProcessingTime(long time, TriggerContext ctx) {
+	public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx)
{
 		return TriggerResult.FIRE_AND_PURGE;
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
index eaca336..cc20296 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java
@@ -53,8 +53,8 @@ public class PurgingTrigger<T, W extends Window> implements Trigger<T,
W> {
 	}
 
 	@Override
-	public TriggerResult onEventTime(long time, TriggerContext ctx) throws Exception {
-		TriggerResult triggerResult = nestedTrigger.onEventTime(time, ctx);
+	public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception
{
+		TriggerResult triggerResult = nestedTrigger.onEventTime(time, window, ctx);
 		switch (triggerResult) {
 			case FIRE:
 				return TriggerResult.FIRE_AND_PURGE;
@@ -66,8 +66,8 @@ public class PurgingTrigger<T, W extends Window> implements Trigger<T,
W> {
 	}
 
 	@Override
-	public TriggerResult onProcessingTime(long time, TriggerContext ctx) throws Exception {
-		TriggerResult triggerResult = nestedTrigger.onProcessingTime(time, ctx);
+	public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception
{
+		TriggerResult triggerResult = nestedTrigger.onProcessingTime(time, window, ctx);
 		switch (triggerResult) {
 			case FIRE:
 				return TriggerResult.FIRE_AND_PURGE;

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
index ef8110b..15ccb33 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java
@@ -60,7 +60,7 @@ public interface Trigger<T, W extends Window> extends Serializable
{
 	 * @param time The timestamp at which the timer fired.
 	 * @param ctx A context object that can be used to register timer callbacks.
 	 */
-	TriggerResult onProcessingTime(long time, TriggerContext ctx) throws Exception;
+	TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;
 
 	/**
 	 * Called when an event-time timer that was set using the trigger context fires.
@@ -68,7 +68,7 @@ public interface Trigger<T, W extends Window> extends Serializable
{
 	 * @param time The timestamp at which the timer fired.
 	 * @param ctx A context object that can be used to register timer callbacks.
 	 */
-	TriggerResult onEventTime(long time, TriggerContext ctx) throws Exception;
+	TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;
 
 
 	/**
@@ -91,19 +91,19 @@ public interface Trigger<T, W extends Window> extends Serializable
{
 
 		/**
 		 * Register a system time callback. When the current system time passes the specified
-		 * time {@link #onProcessingTime(long, TriggerContext)} is called with the time specified
here.
+		 * time {@link #onProcessingTime(long, Window, TriggerContext)} is called with the time
specified here.
 		 *
-		 * @param time The time at which to invoke {@link #onProcessingTime(long, TriggerContext)}
+		 * @param time The time at which to invoke {@link #onProcessingTime(long, Window, TriggerContext)}
 		 */
 		void registerProcessingTimeTimer(long time);
 
 		/**
 		 * Register an event-time callback. When the current watermark passes the specified
-		 * time {@link #onEventTime(long, TriggerContext)} is called with the time specified here.
+		 * time {@link #onEventTime(long, Window, TriggerContext)} is called with the time specified
here.
 		 *
 		 * @see org.apache.flink.streaming.api.watermark.Watermark
 		 *
-		 * @param time The watermark at which to invoke {@link #onEventTime(long, TriggerContext)}
+		 * @param time The watermark at which to invoke {@link #onEventTime(long, Window, TriggerContext)}
 		 */
 		void registerEventTimeTimer(long time);
 

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
index 03e8c4c..2209d5e 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
@@ -437,7 +437,7 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
 
 		public Trigger.TriggerResult onProcessingTime(long time) throws Exception {
 			if (time == processingTimeTimer) {
-				return trigger.onProcessingTime(time, this);
+				return trigger.onProcessingTime(time, window, this);
 			} else {
 				return Trigger.TriggerResult.CONTINUE;
 			}
@@ -445,7 +445,7 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
 
 		public Trigger.TriggerResult onEventTime(long time) throws Exception {
 			if (time == watermarkTimer) {
-				return trigger.onEventTime(time, this);
+				return trigger.onEventTime(time, window, this);
 			} else {
 				return Trigger.TriggerResult.CONTINUE;
 			}

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/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 30ce477..e8e001d 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
@@ -510,7 +510,7 @@ public class WindowOperator<K, IN, OUT, W extends Window>
 
 		public Trigger.TriggerResult onProcessingTime(long time) throws Exception {
 			if (time == processingTimeTimer) {
-				return trigger.onProcessingTime(time, this);
+				return trigger.onProcessingTime(time, window, this);
 			} else {
 				return Trigger.TriggerResult.CONTINUE;
 			}
@@ -518,7 +518,7 @@ public class WindowOperator<K, IN, OUT, W extends Window>
 
 		public Trigger.TriggerResult onEventTime(long time) throws Exception {
 			if (time == watermarkTimer) {
-				return trigger.onEventTime(time, this);
+				return trigger.onEventTime(time, window, this);
 			} else {
 				return Trigger.TriggerResult.CONTINUE;
 			}

http://git-wip-us.apache.org/repos/asf/flink/blob/bc5b852a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
index 0c5cd8f..889ae37 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
@@ -21,7 +21,6 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.junit.Assert;
 
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.LinkedList;
 import java.util.List;


Mime
View raw message