You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by na...@apache.org on 2017/04/19 21:26:08 UTC
samza git commit: SAMZA-1176: Intermittent TestJoinOperator unit test
failure
Repository: samza
Updated Branches:
refs/heads/master 2a92980de -> de8cffdfa
SAMZA-1176: Intermittent TestJoinOperator unit test failure
Join TTL was set too low (10 ms). `joinRetainsMatchedMessagesReverse` will fail if the execution time between line 176 and 186 is longer than that. Increased TTL to 1 min (will not affect test execution time).
Author: Prateek Maheshwari <pm...@linkedin.com>
Reviewers: Navina Ramesh <na...@apache.org>
Closes #131 from prateekm/join-test-fix
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/de8cffdf
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/de8cffdf
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/de8cffdf
Branch: refs/heads/master
Commit: de8cffdfa44a7082a4f7eacfc56c8ef425bc638b
Parents: 2a92980
Author: Prateek Maheshwari <pm...@linkedin.com>
Authored: Wed Apr 19 14:26:00 2017 -0700
Committer: nramesh <nr...@linkedin.com>
Committed: Wed Apr 19 14:26:00 2017 -0700
----------------------------------------------------------------------
.../java/org/apache/samza/operators/TestJoinOperator.java | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/de8cffdf/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java b/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
index 1135726..7a6f959 100644
--- a/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
+++ b/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
@@ -48,6 +48,8 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class TestJoinOperator {
+ private static final Duration JOIN_TTL = Duration.ofMinutes(10);
+
private final TaskCoordinator taskCoordinator = mock(TaskCoordinator.class);
private final Set<Integer> numbers = ImmutableSet.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
@@ -62,7 +64,6 @@ public class TestJoinOperator {
// push messages to second stream with same keys
numbers.forEach(n -> sot.process(new SecondStreamIME(n, n), messageCollector, taskCoordinator));
-
int outputSum = output.stream().reduce(0, (s, m) -> s + m);
assertEquals(110, outputSum);
}
@@ -198,7 +199,7 @@ public class TestJoinOperator {
// push messages to first stream
numbers.forEach(n -> sot.process(new FirstStreamIME(n, n), messageCollector, taskCoordinator));
- testClock.advanceTime(100);
+ testClock.advanceTime(JOIN_TTL.plus(Duration.ofMinutes(1))); // 1 minute after ttl
sot.window(messageCollector, taskCoordinator); // should expire first stream messages
// push messages to second stream with same key
@@ -218,7 +219,7 @@ public class TestJoinOperator {
// push messages to second stream
numbers.forEach(n -> sot.process(new SecondStreamIME(n, n), messageCollector, taskCoordinator));
- testClock.advanceTime(100); // 10 * ttl for join
+ testClock.advanceTime(JOIN_TTL.plus(Duration.ofMinutes(1))); // 1 minute after ttl
sot.window(messageCollector, taskCoordinator); // should expire second stream messages
// push messages to first stream with same key
@@ -254,7 +255,7 @@ public class TestJoinOperator {
SystemStream outputSystemStream = new SystemStream("outputSystem", "outputStream");
inStream
- .join(inStream2, new TestJoinFunction(), Duration.ofMillis(10))
+ .join(inStream2, new TestJoinFunction(), JOIN_TTL)
.sink((message, messageCollector, taskCoordinator) -> {
messageCollector.send(new OutgoingMessageEnvelope(outputSystemStream, message));
});