You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2016/07/21 17:27:38 UTC
[12/13] storm git commit: Unexpected behavior when using count window
together with timestamp extraction
Unexpected behavior when using count window together with timestamp extraction
Do not use timestamp to determine the event count in WatermarkCountEvictionPolicy
when count based trigger is used.
Signed-off-by: P. Taylor Goetz <pt...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/6c813578
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/6c813578
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/6c813578
Branch: refs/heads/1.0.x-branch
Commit: 6c813578c07fa856ad930dbd02d192a752f837a0
Parents: b56a34a
Author: Arun Mahadevan <ar...@apache.org>
Authored: Thu Jul 14 00:00:57 2016 +0530
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Thu Jul 21 13:26:03 2016 -0400
----------------------------------------------------------------------
.../storm/topology/WindowedBoltExecutor.java | 2 +-
.../storm/windowing/CountEvictionPolicy.java | 12 ++--
.../storm/windowing/CountTriggerPolicy.java | 2 +-
.../storm/windowing/DefaultEvictionContext.java | 53 +++++++++++++++
.../apache/storm/windowing/EvictionContext.java | 47 +++++++++++++
.../apache/storm/windowing/EvictionPolicy.java | 3 +-
.../storm/windowing/TimeEvictionPolicy.java | 4 +-
.../storm/windowing/TimeTriggerPolicy.java | 2 +-
.../windowing/WatermarkCountEvictionPolicy.java | 27 +++++---
.../windowing/WatermarkCountTriggerPolicy.java | 2 +-
.../windowing/WatermarkTimeTriggerPolicy.java | 3 +-
.../storm/windowing/WindowManagerTest.java | 69 ++++++++++++++++++--
12 files changed, 197 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java b/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
index 9439ef2..0b18646 100644
--- a/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
@@ -238,7 +238,7 @@ public class WindowedBoltExecutor implements IRichBolt {
WindowManager<Tuple> manager) {
if (windowLengthCount != null) {
if (isTupleTs()) {
- return new WatermarkCountEvictionPolicy<>(windowLengthCount.value, manager);
+ return new WatermarkCountEvictionPolicy<>(windowLengthCount.value);
} else {
return new CountEvictionPolicy<>(windowLengthCount.value);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
index 0e7b233..fb12202 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
@@ -17,7 +17,7 @@
*/
package org.apache.storm.windowing;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
/**
* An eviction policy that tracks event counts and can
@@ -26,12 +26,12 @@ import java.util.concurrent.atomic.AtomicInteger;
* @param <T> the type of event tracked by this policy.
*/
public class CountEvictionPolicy<T> implements EvictionPolicy<T> {
- private final int threshold;
- protected final AtomicInteger currentCount;
+ protected final int threshold;
+ protected final AtomicLong currentCount;
public CountEvictionPolicy(int count) {
this.threshold = count;
- this.currentCount = new AtomicInteger();
+ this.currentCount = new AtomicLong();
}
@Override
@@ -41,7 +41,7 @@ public class CountEvictionPolicy<T> implements EvictionPolicy<T> {
* return if the event should be evicted
*/
while (true) {
- int curVal = currentCount.get();
+ long curVal = currentCount.get();
if (curVal > threshold) {
if (currentCount.compareAndSet(curVal, curVal - 1)) {
return Action.EXPIRE;
@@ -61,7 +61,7 @@ public class CountEvictionPolicy<T> implements EvictionPolicy<T> {
}
@Override
- public void setContext(Object context) {
+ public void setContext(EvictionContext context) {
// NOOP
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/CountTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/CountTriggerPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/CountTriggerPolicy.java
index 2594c67..17750b6 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/CountTriggerPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/CountTriggerPolicy.java
@@ -44,7 +44,7 @@ public class CountTriggerPolicy<T> implements TriggerPolicy<T> {
public void track(Event<T> event) {
if (started && !event.isWatermark()) {
if (currentCount.incrementAndGet() >= count) {
- evictionPolicy.setContext(System.currentTimeMillis());
+ evictionPolicy.setContext(new DefaultEvictionContext(System.currentTimeMillis()));
handler.onTrigger();
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/DefaultEvictionContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/DefaultEvictionContext.java b/storm-core/src/jvm/org/apache/storm/windowing/DefaultEvictionContext.java
new file mode 100644
index 0000000..ef65f66
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/windowing/DefaultEvictionContext.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.windowing;
+
+public class DefaultEvictionContext implements EvictionContext {
+ private final Long referenceTime;
+ private final Long currentCount;
+ private final Long slidingCount;
+
+ public DefaultEvictionContext(Long referenceTime) {
+ this(referenceTime, null);
+ }
+
+ public DefaultEvictionContext(Long referenceTime, Long currentCount) {
+ this(referenceTime, currentCount, null);
+ }
+
+ public DefaultEvictionContext(Long referenceTime, Long currentCount, Long slidingCount) {
+ this.referenceTime = referenceTime;
+ this.currentCount = currentCount;
+ this.slidingCount = slidingCount;
+ }
+
+ @Override
+ public Long getReferenceTime() {
+ return referenceTime;
+ }
+
+ @Override
+ public Long getCurrentCount() {
+ return currentCount;
+ }
+
+ @Override
+ public Long getSlidingCount() {
+ return slidingCount;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/EvictionContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/EvictionContext.java b/storm-core/src/jvm/org/apache/storm/windowing/EvictionContext.java
new file mode 100644
index 0000000..37dcfd9
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/windowing/EvictionContext.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.windowing;
+
+/**
+ * Context information that can be used by the eviction policy
+ */
+public interface EvictionContext {
+ /**
+ * Returns the reference time that the eviction policy could use to
+ * evict the events. In the case of event time processing, this would be
+ * the watermark time.
+ *
+ * @return the reference time in millis
+ */
+ Long getReferenceTime();
+
+ /**
+ * Returns the sliding count for count based windows
+ *
+ * @return the sliding count
+ */
+ Long getSlidingCount();
+
+ /**
+ * Returns the current count of events in the queue up to the reference tim
+ * based on which count based evictions can be performed.
+ *
+ * @return the current count
+ */
+ Long getCurrentCount();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
index 8128d80..05e4d93 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
@@ -70,5 +70,6 @@ public interface EvictionPolicy<T> {
*
* @param context
*/
- void setContext(Object context);
+ void setContext(EvictionContext context);
+
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
index a2db239..e646207 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
@@ -57,8 +57,8 @@ public class TimeEvictionPolicy<T> implements EvictionPolicy<T> {
}
@Override
- public void setContext(Object context) {
- referenceTime = ((Number) context).longValue();
+ public void setContext(EvictionContext context) {
+ referenceTime = context.getReferenceTime();
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/TimeTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/TimeTriggerPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/TimeTriggerPolicy.java
index 1ad7714..b057afb 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/TimeTriggerPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/TimeTriggerPolicy.java
@@ -115,7 +115,7 @@ public class TimeTriggerPolicy<T> implements TriggerPolicy<T> {
* to evict the events
*/
if (evictionPolicy != null) {
- evictionPolicy.setContext(System.currentTimeMillis());
+ evictionPolicy.setContext(new DefaultEvictionContext(System.currentTimeMillis()));
}
handler.onTrigger();
} catch (Throwable th) {
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
index e1af6fb..74240bb 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
@@ -24,25 +24,29 @@ package org.apache.storm.windowing;
* @param <T> the type of event tracked by this policy.
*/
public class WatermarkCountEvictionPolicy<T> extends CountEvictionPolicy<T> {
- private final WindowManager<T> windowManager;
/*
* The reference time in millis for window calculations and
* expiring events. If not set it will default to System.currentTimeMillis()
*/
private long referenceTime;
+ private long processed = 0L;
- public WatermarkCountEvictionPolicy(int count, WindowManager<T> windowManager) {
+ public WatermarkCountEvictionPolicy(int count) {
super(count);
- this.windowManager = windowManager;
}
@Override
public Action evict(Event<T> event) {
- if (event.getTimestamp() <= referenceTime) {
- return super.evict(event);
+ Action action;
+ if (event.getTimestamp() <= referenceTime && processed < currentCount.get()) {
+ action = super.evict(event);
+ if (action == Action.PROCESS) {
+ ++processed;
+ }
} else {
- return Action.KEEP;
+ action = Action.KEEP;
}
+ return action;
}
@Override
@@ -51,9 +55,14 @@ public class WatermarkCountEvictionPolicy<T> extends CountEvictionPolicy<T> {
}
@Override
- public void setContext(Object context) {
- referenceTime = (Long) context;
- currentCount.set(windowManager.getEventCount(referenceTime));
+ public void setContext(EvictionContext context) {
+ referenceTime = context.getReferenceTime();
+ if (context.getCurrentCount() != null) {
+ currentCount.set(context.getCurrentCount());
+ } else {
+ currentCount.set(processed + context.getSlidingCount());
+ }
+ processed = 0;
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountTriggerPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountTriggerPolicy.java
index 391efee..3cfcaad 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountTriggerPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountTriggerPolicy.java
@@ -74,7 +74,7 @@ public class WatermarkCountTriggerPolicy<T> implements TriggerPolicy<T> {
long watermarkTs = waterMarkEvent.getTimestamp();
List<Long> eventTs = windowManager.getSlidingCountTimestamps(lastProcessedTs, watermarkTs, count);
for (long ts : eventTs) {
- evictionPolicy.setContext(ts);
+ evictionPolicy.setContext(new DefaultEvictionContext(ts, null, Long.valueOf(count)));
handler.onTrigger();
lastProcessedTs = ts;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeTriggerPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeTriggerPolicy.java
index 3734c93..00620c4 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeTriggerPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeTriggerPolicy.java
@@ -75,7 +75,8 @@ public class WatermarkTimeTriggerPolicy<T> implements TriggerPolicy<T> {
long windowEndTs = nextWindowEndTs;
LOG.debug("Window end ts {} Watermark ts {}", windowEndTs, watermarkTs);
while (windowEndTs <= watermarkTs) {
- evictionPolicy.setContext(windowEndTs);
+ long currentCount = windowManager.getEventCount(windowEndTs);
+ evictionPolicy.setContext(new DefaultEvictionContext(windowEndTs, currentCount));
if (handler.onTrigger()) {
windowEndTs += slidingIntervalMs;
} else {
http://git-wip-us.apache.org/repos/asf/storm/blob/6c813578/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java b/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
index a545427..6645566 100644
--- a/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
+++ b/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
@@ -187,7 +187,7 @@ public class WindowManagerTest {
windowManager.add(i, now - 1000);
}
// simulate the time trigger by setting the reference time and invoking onTrigger() manually
- evictionPolicy.setContext(now + 100);
+ evictionPolicy.setContext(new DefaultEvictionContext(now + 100));
windowManager.onTrigger();
// 100 events with past ts should expire
@@ -210,7 +210,7 @@ public class WindowManagerTest {
}
activationsEvents.addAll(newEvents);
// simulate the time trigger by setting the reference time and invoking onTrigger() manually
- evictionPolicy.setContext(now + 200);
+ evictionPolicy.setContext(new DefaultEvictionContext(now + 200));
windowManager.onTrigger();
assertTrue(listener.onExpiryEvents.isEmpty());
assertEquals(activationsEvents, listener.onActivationEvents);
@@ -236,20 +236,20 @@ public class WindowManagerTest {
windowManager.add(i);
}
// simulate the time trigger by setting the reference time and invoking onTrigger() manually
- evictionPolicy.setContext(now + 60);
+ evictionPolicy.setContext(new DefaultEvictionContext(now + 60));
windowManager.onTrigger();
assertEquals(seq(1, 10), listener.onActivationEvents);
assertTrue(listener.onActivationExpiredEvents.isEmpty());
listener.clear();
// wait so all events expire
- evictionPolicy.setContext(now + 120);
+ evictionPolicy.setContext(new DefaultEvictionContext(now + 120));
windowManager.onTrigger();
assertEquals(seq(1, 10), listener.onExpiryEvents);
assertTrue(listener.onActivationEvents.isEmpty());
listener.clear();
- evictionPolicy.setContext(now + 180);
+ evictionPolicy.setContext(new DefaultEvictionContext(now + 180));
windowManager.onTrigger();
assertTrue(listener.onActivationExpiredEvents.isEmpty());
assertTrue(listener.onActivationEvents.isEmpty());
@@ -354,7 +354,7 @@ public class WindowManagerTest {
@Test
public void testCountBasedWindowWithEventTs() throws Exception {
- EvictionPolicy<Integer> evictionPolicy = new WatermarkCountEvictionPolicy<>(3, windowManager);
+ EvictionPolicy<Integer> evictionPolicy = new WatermarkCountEvictionPolicy<>(3);
windowManager.setEvictionPolicy(evictionPolicy);
TriggerPolicy<Integer> triggerPolicy = new WatermarkTimeTriggerPolicy<Integer>(10, windowManager, evictionPolicy, windowManager);
triggerPolicy.start();
@@ -430,7 +430,64 @@ public class WindowManagerTest {
assertEquals(seq(9), listener.allOnActivationEvents.get(0));
assertEquals(seq(9, 12), listener.allOnActivationEvents.get(1));
}
+
+ @Test
+ public void testCountBasedTumblingWithSameEventTs() throws Exception {
+ EvictionPolicy<Integer> evictionPolicy = new WatermarkCountEvictionPolicy<>(2);
+ windowManager.setEvictionPolicy(evictionPolicy);
+ TriggerPolicy<Integer> triggerPolicy = new WatermarkCountTriggerPolicy<Integer>(2, windowManager, evictionPolicy, windowManager);
+ triggerPolicy.start();
+ windowManager.setTriggerPolicy(triggerPolicy);
+
+ windowManager.add(1, 10);
+ windowManager.add(2, 10);
+ windowManager.add(3, 11);
+ windowManager.add(4, 12);
+ windowManager.add(5, 12);
+ windowManager.add(6, 12);
+ windowManager.add(7, 12);
+ windowManager.add(8, 13);
+ windowManager.add(9, 14);
+ windowManager.add(10, 15);
+
+ windowManager.add(new WaterMarkEvent<Integer>(20));
+ assertEquals(5, listener.allOnActivationEvents.size());
+ assertEquals(seq(1, 2), listener.allOnActivationEvents.get(0));
+ assertEquals(seq(3, 4), listener.allOnActivationEvents.get(1));
+ assertEquals(seq(5, 6), listener.allOnActivationEvents.get(2));
+ assertEquals(seq(7, 8), listener.allOnActivationEvents.get(3));
+ assertEquals(seq(9, 10), listener.allOnActivationEvents.get(4));
+ }
+
@Test
+ public void testCountBasedSlidingWithSameEventTs() throws Exception {
+ EvictionPolicy<Integer> evictionPolicy = new WatermarkCountEvictionPolicy<>(5);
+ windowManager.setEvictionPolicy(evictionPolicy);
+ TriggerPolicy<Integer> triggerPolicy = new WatermarkCountTriggerPolicy<Integer>(2, windowManager, evictionPolicy, windowManager);
+ triggerPolicy.start();
+ windowManager.setTriggerPolicy(triggerPolicy);
+
+ windowManager.add(1, 10);
+ windowManager.add(2, 10);
+ windowManager.add(3, 11);
+ windowManager.add(4, 12);
+ windowManager.add(5, 12);
+ windowManager.add(6, 12);
+ windowManager.add(7, 12);
+ windowManager.add(8, 13);
+ windowManager.add(9, 14);
+ windowManager.add(10, 15);
+
+ windowManager.add(new WaterMarkEvent<Integer>(20));
+ assertEquals(5, listener.allOnActivationEvents.size());
+ assertEquals(seq(1, 2), listener.allOnActivationEvents.get(0));
+ assertEquals(seq(1, 4), listener.allOnActivationEvents.get(1));
+ assertEquals(seq(2, 6), listener.allOnActivationEvents.get(2));
+ assertEquals(seq(4, 8), listener.allOnActivationEvents.get(3));
+ assertEquals(seq(6, 10), listener.allOnActivationEvents.get(4));
+
+ }
+ @Test
public void testEventTimeLag() throws Exception {
EvictionPolicy<Integer> evictionPolicy = new WatermarkTimeEvictionPolicy<>(20, 5);
windowManager.setEvictionPolicy(evictionPolicy);