Repository: kafka Updated Branches: refs/heads/0.10.0 73e2f090c -> 4091a13c5
hotfix: check join window boundaries Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/9dc5dc46 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/9dc5dc46 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/9dc5dc46 Branch: refs/heads/0.10.0 Commit: 9dc5dc461ea4fcaf4f429cfb6c83b4fd6284921f Parents: 73e2f09 Author: Matthias J. Sax <[email protected]> Authored: Thu Jun 30 01:53:06 2016 +0200 Committer: Matthias J. Sax <[email protected]> Committed: Thu Jun 30 01:53:06 2016 +0200 ---------------------------------------------------------------------- .../kafka/streams/kstream/JoinWindows.java | 21 ++++++++----- .../kafka/streams/kstream/JoinWindowsTest.java | 32 +++++++++++++------- 2 files changed, 34 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/9dc5dc46/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java index 53ddf3e..936bcd2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java @@ -42,7 +42,8 @@ import java.util.Map; * A join is symmetric in the sense, that a join specification on the first stream returns the same result record as * a join specification on the second stream with flipped before and after values. * <p> - * Both values (before and after) must not be negative and not zero at the same time. + * Both values (before and after) must not result in an "inverse" window, + * i.e., lower-interval-bound must not be larger than upper-interval.bound. */ public class JoinWindows extends Windows<TimeWindow> { @@ -54,14 +55,17 @@ public class JoinWindows extends Windows<TimeWindow> { private JoinWindows(String name, long before, long after) { super(name); - if (before < 0) { - throw new IllegalArgumentException("window size must be > 0 (you provided before as " + before + ")"); + if (before < 0) { // shift lower bound to right + if (after < -before) { + throw new IllegalArgumentException("Upper interval bound smaller than lower interval bound." + + " <after> must be at least " + (-before)); + } } - if (after < 0) { - throw new IllegalArgumentException("window size must be > 0 (you provided after as " + after + ")"); - } - if (before == 0 && after == 0) { - throw new IllegalArgumentException("window size must be > 0 (you provided 0)"); + if (after < 0) { // shift upper bound to left + if (before < -after) { + throw new IllegalArgumentException("Lower interval bound greater than upper interval bound." + + " <before> must be at least " + (-after)); + } } this.after = after; @@ -70,6 +74,7 @@ public class JoinWindows extends Windows<TimeWindow> { /** * Specifies that records of the same key are joinable if their timestamps are within {@code timeDifference}. + * ({@code timeDifference} must not be negative) * * @param timeDifference join window interval */ http://git-wip-us.apache.org/repos/asf/kafka/blob/9dc5dc46/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java index d8fa7b4..d80342a 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java @@ -29,7 +29,7 @@ public class JoinWindowsTest { private static String anyName = "window"; private static long anySize = 123L; - private static long anyOtherSize = 456L; + private static long anyOtherSize = 456L; // should be larger than anySize @Test public void shouldHaveSaneEqualsAndHashCode() { @@ -66,6 +66,21 @@ public class JoinWindowsTest { assertNotEquals("must be false when window sizes are different", differentWindowSize3, w1); } + @Test + public void validWindows() { + JoinWindows.of(anyName, anyOtherSize) // [ -anyOtherSize ; anyOtherSize ] + .before(anySize) // [ -anySize ; anyOtherSize ] + .before(0) // [ 0 ; anyOtherSize ] + .before(-anySize) // [ anySize ; anyOtherSize ] + .before(-anyOtherSize); // [ anyOtherSize ; anyOtherSize ] + + JoinWindows.of(anyName, anyOtherSize) // [ -anyOtherSize ; anyOtherSize ] + .after(anySize) // [ -anyOtherSize ; anySize ] + .after(0) // [ -anyOtherSize ; 0 ] + .after(-anySize) // [ -anyOtherSize ; -anySize ] + .after(-anyOtherSize); // [ -anyOtherSize ; -anyOtherSize ] + } + @Test(expected = IllegalArgumentException.class) public void nameMustNotBeEmpty() { JoinWindows.of("", anySize); @@ -77,23 +92,18 @@ public class JoinWindowsTest { } @Test(expected = IllegalArgumentException.class) - public void windowSizeMustNotBeNegative() { + public void timeDifferenceMustNotBeNegative() { JoinWindows.of(anyName, -1); } @Test(expected = IllegalArgumentException.class) - public void beforeMustNotBeNegative() { - JoinWindows.of(anyName, anySize).before(-1); - } - - @Test(expected = IllegalArgumentException.class) - public void afterSizeMustNotBeNegative() { - JoinWindows.of(anyName, anySize).after(-1); + public void afterBelowLower() { + JoinWindows.of(anyName, anySize).after(-anySize-1); } @Test(expected = IllegalArgumentException.class) - public void windowSizeMustNotBeZero() { - JoinWindows.of(anyName, 0); + public void beforeOverUpper() { + JoinWindows.of(anyName, anySize).before(-anySize-1); } } \ No newline at end of file
