You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2015/08/18 15:35:44 UTC
[2/3] flink git commit: [FLINK-2286] [streaming] ITCase for
ParallelMerge behavior
[FLINK-2286] [streaming] ITCase for ParallelMerge behavior
Closes #1014
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/67087dea
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/67087dea
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/67087dea
Branch: refs/heads/master
Commit: 67087dea698881b365debe17f847af0f192ee5cb
Parents: 143ec4f
Author: mbalassi <mb...@apache.org>
Authored: Tue Aug 11 15:51:30 2015 +0200
Committer: mbalassi <mb...@apache.org>
Committed: Tue Aug 18 15:34:18 2015 +0200
----------------------------------------------------------------------
.../graph/test/example/PageRankITCase.java | 2 +-
.../windowing/ParallelMergeITCase.java | 101 +++++++++++++++++++
.../scala/table/test/PageRankTableITCase.java | 2 +-
.../flink/tez/test/PageRankBasicStepITCase.java | 2 +-
.../apache/flink/test/util/TestBaseUtils.java | 11 +-
.../exampleJavaPrograms/PageRankITCase.java | 2 +-
.../exampleScalaPrograms/PageRankITCase.java | 2 +-
7 files changed, 111 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/67087dea/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java
index 544cc66..cde959f 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/PageRankITCase.java
@@ -61,7 +61,7 @@ public class PageRankITCase extends MultipleProgramsTestBase {
@After
public void after() throws Exception{
- compareKeyValueParisWithDelta(expected, resultPath, "\t", 0.01);
+ compareKeyValuePairsWithDelta(expected, resultPath, "\t", 0.01);
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/67087dea/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/operators/windowing/ParallelMergeITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/operators/windowing/ParallelMergeITCase.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/operators/windowing/ParallelMergeITCase.java
new file mode 100644
index 0000000..b762d65
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/operators/windowing/ParallelMergeITCase.java
@@ -0,0 +1,101 @@
+/*
+ * 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.flink.streaming.api.operators.windowing;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.helper.Time;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.util.Collector;
+import org.junit.Assert;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Tests that {@link ParallelMerge} does not swallow records of the
+ * last window.
+ */
+public class ParallelMergeITCase extends StreamingProgramTestBase {
+
+ protected String textPath;
+ protected String resultPath;
+ protected final String input = "To be, or not to be,--that is the question:--" +
+ "Whether 'tis nobler in the mind to suffer";
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", input);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ List<String> resultLines = new ArrayList<>();
+ readAllResultLines(resultLines, resultPath);
+
+ // check that result lines are not swallowed, as every element is expected to be in the
+ // last time window we either get the right output or no output at all
+ if (resultLines.isEmpty()){
+ Assert.fail();
+ }
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> text = env.fromElements(input);
+
+ DataStream<Tuple2<String, Integer>> counts =
+ text.flatMap(new Tokenizer())
+ .window(Time.of(1000, TimeUnit.MILLISECONDS))
+ .groupBy(0)
+ .sum(1)
+ .flatten();
+
+ counts.writeAsText(resultPath);
+
+ try {
+ env.execute();
+ } catch (RuntimeException e){
+ // might happen at closing the active window
+ // do nothing
+ }
+ }
+
+ public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void flatMap(String value, Collector<Tuple2<String, Integer>> out)
+ throws Exception {
+ String[] tokens = value.toLowerCase().split("\\W+");
+
+ for (String token : tokens) {
+ if (token.length() > 0) {
+ out.collect(Tuple2.of(token, 1));
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/67087dea/flink-staging/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java b/flink-staging/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
index 5353b73..1816614 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/scala/table/test/PageRankTableITCase.java
@@ -61,7 +61,7 @@ public class PageRankTableITCase extends JavaProgramTestBase {
@Override
protected void postSubmit() throws Exception {
- compareKeyValueParisWithDelta(expectedResult, resultPath, " ", 0.01);
+ compareKeyValuePairsWithDelta(expectedResult, resultPath, " ", 0.01);
}
@Parameters
http://git-wip-us.apache.org/repos/asf/flink/blob/67087dea/flink-staging/flink-tez/src/test/java/org/apache/flink/tez/test/PageRankBasicStepITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-tez/src/test/java/org/apache/flink/tez/test/PageRankBasicStepITCase.java b/flink-staging/flink-tez/src/test/java/org/apache/flink/tez/test/PageRankBasicStepITCase.java
index 511c2cb..9a203fe 100644
--- a/flink-staging/flink-tez/src/test/java/org/apache/flink/tez/test/PageRankBasicStepITCase.java
+++ b/flink-staging/flink-tez/src/test/java/org/apache/flink/tez/test/PageRankBasicStepITCase.java
@@ -49,6 +49,6 @@ public class PageRankBasicStepITCase extends TezProgramTestBase {
@Override
protected void postSubmit() throws Exception {
- compareKeyValueParisWithDelta(expectedResult, resultPath, " ", 0.001);
+ compareKeyValuePairsWithDelta(expectedResult, resultPath, " ", 0.001);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/67087dea/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
index 9068fcc..c28347c 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
@@ -321,16 +321,15 @@ public class TestBaseUtils extends TestLogger {
Assert.fail(msg);
}
}
-
}
- public static void compareKeyValueParisWithDelta(String expectedLines, String resultPath,
- String delimiter, double maxDelta) throws Exception {
- compareKeyValueParisWithDelta(expectedLines, resultPath, new String[]{}, delimiter, maxDelta);
+ public static void compareKeyValuePairsWithDelta(String expectedLines, String resultPath,
+ String delimiter, double maxDelta) throws Exception {
+ compareKeyValuePairsWithDelta(expectedLines, resultPath, new String[]{}, delimiter, maxDelta);
}
- public static void compareKeyValueParisWithDelta(String expectedLines, String resultPath,
- String[] excludePrefixes, String delimiter, double maxDelta) throws Exception {
+ public static void compareKeyValuePairsWithDelta(String expectedLines, String resultPath,
+ String[] excludePrefixes, String delimiter, double maxDelta) throws Exception {
ArrayList<String> list = new ArrayList<String>();
readAllResultLines(list, resultPath, excludePrefixes, false);
http://git-wip-us.apache.org/repos/asf/flink/blob/67087dea/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java
index 1c66c3e..2d1519d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java
@@ -63,7 +63,7 @@ public class PageRankITCase extends MultipleProgramsTestBase {
@After
public void after() throws Exception{
- compareKeyValueParisWithDelta(expected, resultPath, " ", 0.01);
+ compareKeyValuePairsWithDelta(expected, resultPath, " ", 0.01);
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/67087dea/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java
index f9c2566..6b9e550 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java
@@ -67,7 +67,7 @@ public class PageRankITCase extends MultipleProgramsTestBase {
@After
public void after() throws Exception{
- compareKeyValueParisWithDelta(expected, resultPath, " ", 0.01);
+ compareKeyValuePairsWithDelta(expected, resultPath, " ", 0.01);
}
@Test