You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2014/12/05 18:26:29 UTC
[24/34] incubator-flink git commit: [streaming] Changed time based
windowing policies to make lower boundary excluded and upper boundary
included in the windows and adjusted policy test cases respectively.
[streaming] Changed time based windowing policies to make lower boundary excluded and upper boundary included in the windows and adjusted policy test cases respectively.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/996f02cb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/996f02cb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/996f02cb
Branch: refs/heads/master
Commit: 996f02cbb7a89fde3803a664c9977d7f83f50a40
Parents: 0ef399c
Author: Jonas Traub (powibol) <jo...@s-traub.com>
Authored: Tue Nov 25 12:48:37 2014 +0100
Committer: mbalassi <mb...@apache.org>
Committed: Fri Dec 5 16:45:09 2014 +0100
----------------------------------------------------------------------
.../streaming/api/windowing/policy/TimeEvictionPolicy.java | 2 +-
.../streaming/api/windowing/policy/TimeTriggerPolicy.java | 6 +++---
.../streaming/api/windowing/policy/TimeEvictionPolicyTest.java | 2 +-
.../streaming/api/windowing/policy/TimeTriggerPolicyTest.java | 2 +-
4 files changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/996f02cb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
index e886c09..99116d0 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
@@ -114,7 +114,7 @@ public class TimeEvictionPolicy<DATA> implements ActiveEvictionPolicy<DATA>,
int counter = 0;
while (!buffer.isEmpty()) {
- if (timestamp.getTimestamp(buffer.getFirst()) < threshold) {
+ if (timestamp.getTimestamp(buffer.getFirst()) <= threshold) {
buffer.removeFirst();
counter++;
} else {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/996f02cb/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
index c3402c9..3539ad6 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
@@ -109,7 +109,7 @@ public class TimeTriggerPolicy<DATA> implements ActiveTriggerPolicy<DATA>,
// use > here. In case >= would fit, the regular call will do the job.
while (timestamp.getTimestamp(datapoint) > startTime + granularity) {
startTime += granularity;
- fakeElements.add(startTime);
+ fakeElements.add(startTime-1);
}
return (Object[]) fakeElements.toArray();
}
@@ -143,9 +143,10 @@ public class TimeTriggerPolicy<DATA> implements ActiveTriggerPolicy<DATA>,
*/
private synchronized void activeFakeElementEmission(ActiveTriggerCallback callback) {
+ // start time is excluded, but end time is included: >=
if (System.currentTimeMillis() >= startTime + granularity) {
startTime += granularity;
- callback.sendFakeElement(startTime);
+ callback.sendFakeElement(startTime-1);
}
}
@@ -177,7 +178,6 @@ public class TimeTriggerPolicy<DATA> implements ActiveTriggerPolicy<DATA>,
@Override
public synchronized boolean notifyTrigger(DATA datapoint) {
long recordTime = timestamp.getTimestamp(datapoint);
- // start time is included, but end time is excluded: >=
if (recordTime >= startTime + granularity) {
if (granularity != 0) {
startTime = recordTime - ((recordTime - startTime) % granularity);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/996f02cb/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
index 303c66a..82c8841 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
@@ -89,7 +89,7 @@ public class TimeEvictionPolicyTest {
// handle correctness of eviction
for (; result > 0 && !buffer.isEmpty(); result--) {
- if (buffer.getFirst() < times[i] - granularity) {
+ if (buffer.getFirst() <= times[i] - granularity) {
buffer.removeFirst();
} else {
fail("The policy wanted to evict time " + buffer.getFirst()
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/996f02cb/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
index 1355b27..28d35ec 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
@@ -105,7 +105,7 @@ public class TimeTriggerPolicyTest {
timeStamp);
// expected result
- Long[][] result = { {}, {}, { 5L, 10L, 15L }, { 25L } };
+ Long[][] result = { {}, {}, { 4L, 9L, 14L }, { 24L } };
// call policy
for (int i = 0; i < times.length; i++) {