You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2016/02/04 16:51:58 UTC
flink git commit: [FLINK-2845] Fix
TimestampITCase.testWatermarkPropagation()
Repository: flink
Updated Branches:
refs/heads/master f8f747f22 -> 69f7f6d92
[FLINK-2845] Fix TimestampITCase.testWatermarkPropagation()
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/69f7f6d9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/69f7f6d9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/69f7f6d9
Branch: refs/heads/master
Commit: 69f7f6d92f263e7fa73fe5e396565aa87d283818
Parents: f8f747f
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Feb 2 10:49:01 2016 +0100
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Thu Feb 4 16:44:12 2016 +0100
----------------------------------------------------------------------
.../streaming/timestamp/TimestampITCase.java | 56 ++++++--------------
1 file changed, 17 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/69f7f6d9/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
index 4c922a7..0e03b71 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
@@ -47,6 +47,7 @@ import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
@@ -141,42 +142,11 @@ public class TimestampITCase {
// verify that all the watermarks arrived at the final custom operator
for (int i = 0; i < PARALLELISM; i++) {
- // There can be two cases, either we get NUM_WATERMARKS + 1 watermarks or
- // (NUM_WATERMARKS / 2) + 1 watermarks. This depends on which source get's to run first.
- // If source1 runs first we jump directly to +Inf and skip all the intermediate
- // watermarks. If source2 runs first we see the intermediate watermarks from
- // NUM_WATERMARKS/2 to +Inf.
- if (CustomOperator.finalWatermarks[i].size() == NUM_WATERMARKS + 1) {
- for (int j = 0; j < NUM_WATERMARKS; j++) {
- if (!CustomOperator.finalWatermarks[i].get(j).equals(new Watermark(initialTime + j))) {
- System.err.println("All Watermarks: ");
- for (int k = 0; k <= NUM_WATERMARKS; k++) {
- System.err.println(CustomOperator.finalWatermarks[i].get(k));
- }
-
- Assert.fail("Wrong watermark.");
- }
- }
- if (!CustomOperator.finalWatermarks[i].get(NUM_WATERMARKS).equals(new Watermark(Long.MAX_VALUE))) {
- System.err.println("All Watermarks: ");
- for (int k = 0; k <= NUM_WATERMARKS; k++) {
- System.err.println(CustomOperator.finalWatermarks[i].get(k));
- }
-
- Assert.fail("Wrong watermark.");
- }
- } else {
- for (int j = 0; j < NUM_WATERMARKS / 2; j++) {
- if (!CustomOperator.finalWatermarks[i].get(j).equals(new Watermark(initialTime + j))) {
- System.err.println("All Watermarks: ");
- for (int k = 0; k <= NUM_WATERMARKS / 2; k++) {
- System.err.println(CustomOperator.finalWatermarks[i].get(k));
- }
-
- Assert.fail("Wrong watermark.");
- }
- }
- if (!CustomOperator.finalWatermarks[i].get(NUM_WATERMARKS / 2).equals(new Watermark(Long.MAX_VALUE))) {
+ // we are only guaranteed to see NUM_WATERMARKS / 2 watermarks in order, because
+ // after that source2 emits Long.MAX_VALUE which could match with an arbitrary
+ // mark from source 1, for example, we could see 0,1,2,4,5,7,MAX
+ for (int j = 0; j < NUM_WATERMARKS / 2; j++) {
+ if (!CustomOperator.finalWatermarks[i].get(j).equals(new Watermark(initialTime + j))) {
System.err.println("All Watermarks: ");
for (int k = 0; k <= NUM_WATERMARKS / 2; k++) {
System.err.println(CustomOperator.finalWatermarks[i].get(k));
@@ -184,9 +154,14 @@ public class TimestampITCase {
Assert.fail("Wrong watermark.");
}
-
}
-
+ if (!CustomOperator.finalWatermarks[i].get(CustomOperator.finalWatermarks[i].size() - 1).equals(new Watermark(Long.MAX_VALUE))) {
+ System.err.println("All Watermarks: ");
+ for (int k = 0; k <= NUM_WATERMARKS; k++) {
+ System.err.println(CustomOperator.finalWatermarks[i].get(k));
+ }
+ Assert.fail("Wrong watermark.");
+ }
}
}
@@ -615,6 +590,9 @@ public class TimestampITCase {
@Override
public void processWatermark(Watermark mark) throws Exception {
+ for (Watermark previousMark: watermarks) {
+ assertTrue(previousMark.getTimestamp() < mark.getTimestamp());
+ }
watermarks.add(mark);
latch.trigger();
output.emitWatermark(mark);
@@ -623,7 +601,7 @@ public class TimestampITCase {
@Override
public void open() throws Exception {
super.open();
- watermarks = new ArrayList<Watermark>();
+ watermarks = new ArrayList<>();
}
@Override