You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/01/13 01:41:17 UTC
[1/7] storm git commit: STORM-1841: Address a few minor issues in
windowing and doc
Repository: storm
Updated Branches:
refs/heads/1.0.x-branch f49959cf0 -> 75dac521a
STORM-1841: Address a few minor issues in windowing and doc
1. Do not accept negative values for window length or sliding interval in BaseWindowedBolt
2. Added static factories for Count and Duration for ease of use
3. Explicitly call out the sliding behavior and when the first window is evaluated in the windowing doc.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9f453ad5
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9f453ad5
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9f453ad5
Branch: refs/heads/1.0.x-branch
Commit: 9f453ad50e2b88af0a1646c788a68074d708811d
Parents: f49959c
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Mon May 16 01:44:41 2016 +0530
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jan 13 10:27:01 2017 +0900
----------------------------------------------------------------------
docs/Windowing.md | 13 +--
.../storm/starter/SlidingWindowTopology.java | 4 +-
.../storm/topology/base/BaseWindowedBolt.java | 87 ++++++++++++++++++++
3 files changed, 96 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9f453ad5/docs/Windowing.md
----------------------------------------------------------------------
diff --git a/docs/Windowing.md b/docs/Windowing.md
index 44512f7..9f5869f 100644
--- a/docs/Windowing.md
+++ b/docs/Windowing.md
@@ -17,15 +17,16 @@ Tuples are grouped in windows and window slides every sliding interval. A tuple
For example a time duration based sliding window with length 10 secs and sliding interval of 5 seconds.
```
-| e1 e2 | e3 e4 e5 e6 | e7 e8 e9 |...
-0 5 10 15 -> time
-
-|<------- w1 -------->|
- |------------ w2 ------->|
+........| e1 e2 | e3 e4 e5 e6 | e7 e8 e9 |...
+-5 0 5 10 15 -> time
+|<------- w1 -->|
+ |<---------- w2 ----->|
+ |<-------------- w3 ---->|
```
The window is evaluated every 5 seconds and some of the tuples in the first window overlaps with the second one.
-
+
+Note: The window first slides at t = 5 secs and would contain events received up to the first five secs.
## Tumbling Window
http://git-wip-us.apache.org/repos/asf/storm/blob/9f453ad5/examples/storm-starter/src/jvm/org/apache/storm/starter/SlidingWindowTopology.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/org/apache/storm/starter/SlidingWindowTopology.java b/examples/storm-starter/src/jvm/org/apache/storm/starter/SlidingWindowTopology.java
index cedcec5..8477ad1 100644
--- a/examples/storm-starter/src/jvm/org/apache/storm/starter/SlidingWindowTopology.java
+++ b/examples/storm-starter/src/jvm/org/apache/storm/starter/SlidingWindowTopology.java
@@ -87,9 +87,9 @@ public class SlidingWindowTopology {
public static void main(String[] args) throws Exception {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("integer", new RandomIntegerSpout(), 1);
- builder.setBolt("slidingsum", new SlidingWindowSumBolt().withWindow(new Count(30), new Count(10)), 1)
+ builder.setBolt("slidingsum", new SlidingWindowSumBolt().withWindow(Count.of(30), Count.of(10)), 1)
.shuffleGrouping("integer");
- builder.setBolt("tumblingavg", new TumblingWindowAvgBolt().withTumblingWindow(new Count(3)), 1)
+ builder.setBolt("tumblingavg", new TumblingWindowAvgBolt().withTumblingWindow(Count.of(3)), 1)
.shuffleGrouping("slidingsum");
builder.setBolt("printer", new PrinterBolt(), 1).shuffleGrouping("tumblingavg");
Config conf = new Config();
http://git-wip-us.apache.org/repos/asf/storm/blob/9f453ad5/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java b/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
index a1d9ebe..e1725dd 100644
--- a/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
+++ b/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
@@ -43,6 +43,23 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
public Count(int value) {
this.value = value;
}
+
+ /**
+ * Returns a {@link Count} of given value.
+ *
+ * @param value the count value
+ * @return the Count
+ */
+ public static Count of(int value) {
+ return new Count(value);
+ }
+
+ @Override
+ public String toString() {
+ return "Count{" +
+ "value=" + value +
+ '}';
+ }
}
/**
@@ -54,6 +71,63 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
public Duration(int value, TimeUnit timeUnit) {
this.value = (int) timeUnit.toMillis(value);
}
+
+ /**
+ * Returns a {@link Duration} corresponding to the the given value in milli seconds.
+ *
+ * @param milliseconds the duration in milliseconds
+ * @return the Duration
+ */
+ public static Duration of(int milliseconds) {
+ return new Duration(milliseconds, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Returns a {@link Duration} corresponding to the the given value in days.
+ *
+ * @param days the number of days
+ * @return the Duration
+ */
+ public static Duration days(int days) {
+ return new Duration(days, TimeUnit.DAYS);
+ }
+
+ /**
+ * Returns a {@link Duration} corresponding to the the given value in hours.
+ *
+ * @param hours the number of hours
+ * @return the Duration
+ */
+ public static Duration hours(int hours) {
+ return new Duration(hours, TimeUnit.HOURS);
+ }
+
+ /**
+ * Returns a {@link Duration} corresponding to the the given value in minutes.
+ *
+ * @param minutes the number of minutes
+ * @return the Duration
+ */
+ public static Duration minutes(int minutes) {
+ return new Duration(minutes, TimeUnit.MINUTES);
+ }
+
+ /**
+ * Returns a {@link Duration} corresponding to the the given value in seconds.
+ *
+ * @param seconds the number of seconds
+ * @return the Duration
+ */
+ public static Duration seconds(int seconds) {
+ return new Duration(seconds, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public String toString() {
+ return "Duration{" +
+ "value=" + value +
+ '}';
+ }
}
protected BaseWindowedBolt() {
@@ -61,21 +135,34 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
}
private BaseWindowedBolt withWindowLength(Count count) {
+ if (count.value < 0) {
+ throw new IllegalArgumentException("Negative window length [" + count + "]");
+ }
windowConfiguration.put(Config.TOPOLOGY_BOLTS_WINDOW_LENGTH_COUNT, count.value);
return this;
}
private BaseWindowedBolt withWindowLength(Duration duration) {
+ if (duration.value < 0) {
+ throw new IllegalArgumentException("Negative window length [" + duration + "]");
+ }
+
windowConfiguration.put(Config.TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS, duration.value);
return this;
}
private BaseWindowedBolt withSlidingInterval(Count count) {
+ if (count.value < 0) {
+ throw new IllegalArgumentException("Negative sliding interval [" + count + "]");
+ }
windowConfiguration.put(Config.TOPOLOGY_BOLTS_SLIDING_INTERVAL_COUNT, count.value);
return this;
}
private BaseWindowedBolt withSlidingInterval(Duration duration) {
+ if (duration.value < 0) {
+ throw new IllegalArgumentException("Negative sliding interval [" + duration + "]");
+ }
windowConfiguration.put(Config.TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS, duration.value);
return this;
}
[5/7] storm git commit: STORM-2246: Logviewer download link has
urlencoding on part of the URL
Posted by ka...@apache.org.
STORM-2246: Logviewer download link has urlencoding on part of the URL
Changed the URL format for "Download full file" link under worker logs
from "download/URLENCODED(<filename>)" to "download?file=URLENCODED(<filename>)"
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2b7af90f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2b7af90f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2b7af90f
Branch: refs/heads/1.0.x-branch
Commit: 2b7af90f554a647d7b943bd2456ba661c048e28a
Parents: 10147d6
Author: Arun Mahadevan <ar...@apache.org>
Authored: Tue Dec 13 16:23:35 2016 +0530
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jan 13 10:30:10 2017 +0900
----------------------------------------------------------------------
storm-core/src/clj/org/apache/storm/daemon/logviewer.clj | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/2b7af90f/storm-core/src/clj/org/apache/storm/daemon/logviewer.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/daemon/logviewer.clj b/storm-core/src/clj/org/apache/storm/daemon/logviewer.clj
index 92224c6..9b96c86 100644
--- a/storm-core/src/clj/org/apache/storm/daemon/logviewer.clj
+++ b/storm-core/src/clj/org/apache/storm/daemon/logviewer.clj
@@ -396,7 +396,7 @@
"Next" :enabled (> next-start start))])]]))
(defn- download-link [fname]
- [[:p (link-to (url-format "/download/%s" fname) "Download Full File")]])
+ [[:p (link-to (url-format "/download?file=%s" (to-array [fname])) "Download Full File")]])
(defn- daemon-download-link [fname]
[[:p (link-to (url-format "/daemondownload/%s" fname) "Download Full File")]])
@@ -1086,10 +1086,11 @@
(catch InvalidRequestException ex
(log-error ex)
(ring-response-from-exception ex))))
- (GET "/download/:file" [:as {:keys [servlet-request servlet-response log-root]} file & m]
+ (GET "/download" [:as {:keys [servlet-request servlet-response log-root]} & m]
(try
- (mark! logviewer:num-download-log-file-http-requests)
- (let [user (.getUserName http-creds-handler servlet-request)]
+ (.mark logviewer:num-download-log-file-http-requests)
+ (let [user (.getUserName http-creds-handler servlet-request)
+ file (url-decode (:file m))]
(download-log-file file servlet-request servlet-response user log-root))
(catch InvalidRequestException ex
(log-error ex)
[7/7] storm git commit: STORM-2246, STORM-1906,
STORM-1841: CHANGELOG (backport)
Posted by ka...@apache.org.
STORM-2246, STORM-1906, STORM-1841: CHANGELOG (backport)
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/75dac521
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/75dac521
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/75dac521
Branch: refs/heads/1.0.x-branch
Commit: 75dac521a19fb68b73c2e61f0e1060ad9b0e99fa
Parents: f2ad46a
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Jan 13 10:40:55 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jan 13 10:40:55 2017 +0900
----------------------------------------------------------------------
CHANGELOG.md | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/75dac521/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b11c48f..413e75c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,7 @@
## 1.0.3
+ * STORM-2246: Logviewer download link has urlencoding on part of the URL
+ * STORM-1906: Window count/length of zero should be disallowed
+ * STORM-1841: Address a few minor issues in windowing and doc
* STORM-2268: Fix integration test for Travis CI build
* STORM-2283: Fix DefaultStateHandler kryo multithreading issues
* STORM-2279: Unable to open bolt page of storm ui
[2/7] storm git commit: Merge branch 'STORM-1841-1.0.x-merge' into
1.0.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-1841-1.0.x-merge' into 1.0.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/fc4e15f2
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/fc4e15f2
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/fc4e15f2
Branch: refs/heads/1.0.x-branch
Commit: fc4e15f2bdfb761677b1bdbdf0924a8ecff6a16b
Parents: f49959c 9f453ad
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Jan 13 10:27:31 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jan 13 10:27:31 2017 +0900
----------------------------------------------------------------------
docs/Windowing.md | 13 +--
.../storm/starter/SlidingWindowTopology.java | 4 +-
.../storm/topology/base/BaseWindowedBolt.java | 87 ++++++++++++++++++++
3 files changed, 96 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
[4/7] storm git commit: Merge branch 'STORM-1906-1.0.x-merge' into
1.0.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-1906-1.0.x-merge' into 1.0.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/10147d65
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/10147d65
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/10147d65
Branch: refs/heads/1.0.x-branch
Commit: 10147d6508d2c16378937c9f1ccce340f6a4c056
Parents: fc4e15f 42b675d
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Jan 13 10:27:55 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jan 13 10:27:55 2017 +0900
----------------------------------------------------------------------
.../storm/topology/base/BaseWindowedBolt.java | 16 ++++++++--------
1 file changed, 8 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
[3/7] storm git commit: STORM-1906: Window count/length of zero
should be disallowed
Posted by ka...@apache.org.
STORM-1906: Window count/length of zero should be disallowed
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/42b675d3
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/42b675d3
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/42b675d3
Branch: refs/heads/1.0.x-branch
Commit: 42b675d3f5491d6b12f709a3c84421f7f6d5f1c2
Parents: fc4e15f
Author: Raghav Kumar Gautam <ra...@apache.org>
Authored: Wed Jun 15 12:35:37 2016 -0700
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jan 13 10:27:50 2017 +0900
----------------------------------------------------------------------
.../storm/topology/base/BaseWindowedBolt.java | 16 ++++++++--------
1 file changed, 8 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/42b675d3/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java b/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
index e1725dd..561008a 100644
--- a/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
+++ b/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
@@ -135,16 +135,16 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
}
private BaseWindowedBolt withWindowLength(Count count) {
- if (count.value < 0) {
- throw new IllegalArgumentException("Negative window length [" + count + "]");
+ if (count.value <= 0) {
+ throw new IllegalArgumentException("Window length must be positive [" + count + "]");
}
windowConfiguration.put(Config.TOPOLOGY_BOLTS_WINDOW_LENGTH_COUNT, count.value);
return this;
}
private BaseWindowedBolt withWindowLength(Duration duration) {
- if (duration.value < 0) {
- throw new IllegalArgumentException("Negative window length [" + duration + "]");
+ if (duration.value <= 0) {
+ throw new IllegalArgumentException("Window length must be positive [" + duration + "]");
}
windowConfiguration.put(Config.TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS, duration.value);
@@ -152,16 +152,16 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
}
private BaseWindowedBolt withSlidingInterval(Count count) {
- if (count.value < 0) {
- throw new IllegalArgumentException("Negative sliding interval [" + count + "]");
+ if (count.value <= 0) {
+ throw new IllegalArgumentException("Sliding interval must be positive [" + count + "]");
}
windowConfiguration.put(Config.TOPOLOGY_BOLTS_SLIDING_INTERVAL_COUNT, count.value);
return this;
}
private BaseWindowedBolt withSlidingInterval(Duration duration) {
- if (duration.value < 0) {
- throw new IllegalArgumentException("Negative sliding interval [" + duration + "]");
+ if (duration.value <= 0) {
+ throw new IllegalArgumentException("Sliding interval must be positive [" + duration + "]");
}
windowConfiguration.put(Config.TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS, duration.value);
return this;
[6/7] storm git commit: Merge branch 'STORM-2246-1.0.x-merge' into
1.0.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-2246-1.0.x-merge' into 1.0.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f2ad46a9
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f2ad46a9
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f2ad46a9
Branch: refs/heads/1.0.x-branch
Commit: f2ad46a973c9ae79625c1135820d4e8aecb49241
Parents: 10147d6 2b7af90
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Jan 13 10:30:15 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jan 13 10:30:15 2017 +0900
----------------------------------------------------------------------
storm-core/src/clj/org/apache/storm/daemon/logviewer.clj | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------