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/02/19 11:42:31 UTC
[1/2] flink git commit: [FLINK-1560][streaming] streaming WordCount
ITCases
Repository: flink
Updated Branches:
refs/heads/master c1a334e1c -> a673e7ba6
[FLINK-1560][streaming] streaming WordCount ITCases
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a673e7ba
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a673e7ba
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a673e7ba
Branch: refs/heads/master
Commit: a673e7ba646254c6946a6cb660c4dd9dd71959f5
Parents: 6049398
Author: mbalassi <mb...@apache.org>
Authored: Wed Feb 18 16:19:58 2015 +0100
Committer: mbalassi <mb...@apache.org>
Committed: Thu Feb 19 10:40:34 2015 +0100
----------------------------------------------------------------------
.../flink-streaming-examples/pom.xml | 35 +++++-
.../examples/wordcount/PojoExample.java | 2 +-
.../streaming/examples/wordcount/WordCount.java | 16 +--
.../examples/test/PojoExampleITCase.java | 45 ++++++++
.../examples/test/WordCountITCase.java | 45 ++++++++
.../flink/test/testdata/WordCountData.java | 108 ++++++++++++++++++-
6 files changed, 237 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a673e7ba/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/pom.xml b/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
index 6a5ae36..a847284 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
+++ b/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
@@ -58,6 +58,28 @@ under the License.
<artifactId>flink-streaming-connectors</artifactId>
<version>${project.version}</version>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <classifier>tests</classifier>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-tests</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
@@ -437,6 +459,17 @@ under the License.
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
<pluginManagement>
@@ -468,7 +501,7 @@ under the License.
</plugin>
</plugins>
</pluginManagement>
-
+
</build>
</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/a673e7ba/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
index f377863..d582c10 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
@@ -114,7 +114,7 @@ public class PojoExample {
@Override
public String toString() {
- return "(" + word + ", " + frequency + ")";
+ return "(" + word + "," + frequency + ")";
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a673e7ba/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
index 085fe5f..c207d60 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
@@ -17,8 +17,6 @@
package org.apache.flink.streaming.examples.wordcount;
-import java.util.StringTokenizer;
-
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.examples.java.wordcount.util.WordCountData;
@@ -73,7 +71,7 @@ public class WordCount {
// emit result
if (fileOutput) {
- counts.writeAsText(outputPath, 1);
+ counts.writeAsText(outputPath);
} else {
counts.print();
}
@@ -96,14 +94,16 @@ public class WordCount {
private static final long serialVersionUID = 1L;
@Override
- public void flatMap(String inTuple, Collector<Tuple2<String, Integer>> out)
+ public void flatMap(String value, Collector<Tuple2<String, Integer>> out)
throws Exception {
- // tokenize the line
- StringTokenizer tokenizer = new StringTokenizer(inTuple);
+ // normalize and split the line
+ String[] tokens = value.toLowerCase().split("\\W+");
// emit the pairs
- while (tokenizer.hasMoreTokens()) {
- out.collect(new Tuple2<String, Integer>(tokenizer.nextToken(), 1));
+ for (String token : tokens) {
+ if (token.length() > 0) {
+ out.collect(new Tuple2<String, Integer>(token, 1));
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a673e7ba/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/PojoExampleITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/PojoExampleITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/PojoExampleITCase.java
new file mode 100644
index 0000000..b367ff7
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/PojoExampleITCase.java
@@ -0,0 +1,45 @@
+/*
+ * 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.examples.test;
+
+import org.apache.flink.streaming.examples.wordcount.PojoExample;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.test.testdata.WordCountData;
+
+public class PojoExampleITCase extends StreamingProgramTestBase {
+
+ protected String textPath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.STREAMING_COUNTS_AS_TUPLES, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ PojoExample.main(new String[]{textPath, resultPath});
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a673e7ba/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/WordCountITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/WordCountITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/WordCountITCase.java
new file mode 100644
index 0000000..3aba60c
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/examples/test/WordCountITCase.java
@@ -0,0 +1,45 @@
+/*
+ * 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.examples.test;
+
+import org.apache.flink.streaming.examples.wordcount.WordCount;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+
+public class WordCountITCase extends StreamingProgramTestBase {
+
+ protected String textPath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.STREAMING_COUNTS_AS_TUPLES, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ WordCount.main(new String[] { textPath, resultPath });
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a673e7ba/flink-test-utils/src/main/java/org/apache/flink/test/testdata/WordCountData.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/testdata/WordCountData.java b/flink-test-utils/src/main/java/org/apache/flink/test/testdata/WordCountData.java
index e70360b..23c649b 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/testdata/WordCountData.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/testdata/WordCountData.java
@@ -159,8 +159,107 @@ public class WordCountData {
+ "verboten 1\n" + "laeg 1\n" + "nie 1\n" + "drei 2\n" + "dauern 1\n" + "toren 1\n" + "dauert 1\n"
+ "verheeren 1\n" + "fliegend 1\n" + "aus 1\n" + "staub 1\n" + "fluessen 1\n" + "haus 1\n" + "auf 5\n"
+ "dient 2\n" + "tiefer 1\n" + "naeh 1\n" + "zieren 1\n";
-
-
+
+ public static final String STREAMING_COUNTS_AS_TUPLES = "(machen,1)\n" + "(zeit,1)\n" + "(zeit,2)\n" + "(heerscharen,1)\n" + "(keiner,1)\n"
+ + "(keiner,2)\n" + "(meine,1)\n" + "(meine,2)\n" + "(meine,3)\n" + "(fuehr,1)\n" + "(triumph,1)\n" + "(kommst,1)\n"
+ + "(frei,1)\n" + "(schaffen,1)\n" + "(gesinde,1)\n" + "(langbeinigen,1)\n" + "(schalk,1)\n" + "(besser,1)\n" + "(solang,1)\n"
+ + "(meer,1)\n" + "(meer,2)\n" + "(meer,3)\n" + "(meer,4)\n" + "(fragst,1)\n" + "(gabriel,1)\n" + "(selbst,1)\n" + "(selbst,2)\n"
+ + "(bin,1)\n" + "(sich,1)\n" + "(sich,2)\n" + "(sich,3)\n" + "(sich,4)\n" + "(sich,5)\n" + "(sich,6)\n" + "(sich,7)\n" + "(du,1)\n"
+ + "(du,2)\n" + "(du,3)\n" + "(du,4)\n" + "(du,5)\n" + "(du,6)\n" + "(du,7)\n" + "(du,8)\n" + "(du,9)\n" + "(du,10)\n" + "(du,11)\n"
+ + "(sogar,1)\n" + "(geht,1)\n" + "(immer,1)\n" + "(immer,2)\n" + "(immer,3)\n" + "(immer,4)\n" + "(mensch,1)\n" + "(mensch,2)\n"
+ + "(befestigt,1)\n" + "(lebt,1)\n" + "(lebt,2)\n" + "(mag,1)\n" + "(mag,2)\n" + "(mag,3)\n" + "(engeln,1)\n" + "(engeln,2)\n"
+ + "(breiten,1)\n" + "(blitzendes,1)\n" + "(tags,1)\n" + "(sie,1)\n" + "(sie,2)\n" + "(plagen,1)\n" + "(plagen,2)\n" + "(allzu,1)\n"
+ + "(meisten,1)\n" + "(o,1)\n" + "(pfade,1)\n" + "(kennst,1)\n" + "(nichts,1)\n" + "(nichts,2)\n" + "(nichts,3)\n" + "(gedanken,1)\n"
+ + "(befriedigt,1)\n" + "(mich,1)\n" + "(mich,2)\n" + "(mich,3)\n" + "(mich,4)\n" + "(mich,5)\n" + "(mich,6)\n" + "(s,1)\n" + "(s,2)\n"
+ + "(s,3)\n" + "(es,1)\n" + "(es,2)\n" + "(es,3)\n" + "(es,4)\n" + "(es,5)\n" + "(es,6)\n" + "(es,7)\n" + "(es,8)\n" + "(verneinen,1)\n"
+ + "(er,1)\n" + "(er,2)\n" + "(er,3)\n" + "(er,4)\n" + "(er,5)\n" + "(er,6)\n" + "(er,7)\n" + "(er,8)\n" + "(er,9)\n" + "(er,10)\n"
+ + "(er,11)\n" + "(er,12)\n" + "(er,13)\n" + "(gleich,1)\n" + "(baeumchen,1)\n" + "(donnergang,1)\n" + "(wunderlich,1)\n"
+ + "(reise,1)\n" + "(urquell,1)\n" + "(doch,1)\n" + "(doch,2)\n" + "(doch,3)\n" + "(aufs,1)\n" + "(aufs,2)\n" + "(toten,1)\n"
+ + "(niemals,1)\n" + "(eine,1)\n" + "(eine,2)\n" + "(hab,1)\n" + "(darfst,1)\n" + "(da,1)\n" + "(da,2)\n" + "(da,3)\n" + "(da,4)\n"
+ + "(da,5)\n" + "(gen,1)\n" + "(einem,1)\n" + "(einem,2)\n" + "(teil,1)\n" + "(das,1)\n" + "(das,2)\n" + "(das,3)\n" + "(das,4)\n"
+ + "(das,5)\n" + "(das,6)\n" + "(das,7)\n" + "(speise,1)\n" + "(wenig,1)\n" + "(sterne,1)\n" + "(geb,1)\n" + "(welten,1)\n"
+ + "(alle,1)\n" + "(alle,2)\n" + "(alle,3)\n" + "(toent,1)\n" + "(gras,1)\n" + "(felsen,1)\n" + "(kette,1)\n" + "(ich,1)\n"
+ + "(ich,2)\n" + "(ich,3)\n" + "(ich,4)\n" + "(ich,5)\n" + "(ich,6)\n" + "(ich,7)\n" + "(ich,8)\n" + "(ich,9)\n" + "(ich,10)\n"
+ + "(ich,11)\n" + "(ich,12)\n" + "(ich,13)\n" + "(ich,14)\n" + "(fuer,1)\n" + "(fuer,2)\n" + "(als,1)\n" + "(als,2)\n" + "(als,3)\n"
+ + "(mein,1)\n" + "(schoene,1)\n" + "(verzeih,1)\n" + "(schwankender,1)\n" + "(wie,1)\n" + "(wie,2)\n" + "(wie,3)\n" + "(wie,4)\n"
+ + "(wie,5)\n" + "(wie,6)\n" + "(wie,7)\n" + "(wie,8)\n" + "(wie,9)\n" + "(menschlich,1)\n" + "(gaertner,1)\n" + "(taetigkeit,1)\n"
+ + "(bange,1)\n" + "(liebe,1)\n" + "(sei,1)\n" + "(sei,2)\n" + "(seh,1)\n" + "(tollheit,1)\n" + "(am,1)\n" + "(am,2)\n" + "(am,3)\n"
+ + "(am,4)\n" + "(am,5)\n" + "(am,6)\n" + "(michael,1)\n" + "(geist,1)\n" + "(ab,1)\n" + "(nahst,1)\n" + "(vollendet,1)\n"
+ + "(liebt,1)\n" + "(brausen,1)\n" + "(nase,1)\n" + "(erlaubt,1)\n" + "(weiss,1)\n" + "(weiss,2)\n" + "(schnellem,1)\n"
+ + "(deinem,1)\n" + "(gleichem,1)\n" + "(gaerung,1)\n" + "(dauernden,1)\n" + "(deines,1)\n" + "(vorgeschriebne,1)\n"
+ + "(irdisch,1)\n" + "(worte,1)\n" + "(verehren,1)\n" + "(hohen,1)\n" + "(hohen,2)\n" + "(weise,1)\n" + "(weise,2)\n"
+ + "(kuenft,1)\n" + "(werdende,1)\n" + "(wette,1)\n" + "(wette,2)\n" + "(wuetend,1)\n" + "(erscheinung,1)\n" + "(gar,1)\n"
+ + "(gar,2)\n" + "(verlieren,1)\n" + "(braucht,1)\n" + "(weiter,1)\n" + "(trank,1)\n" + "(tierischer,1)\n" + "(wohl,1)\n"
+ + "(verteilen,1)\n" + "(verhoehnt,1)\n" + "(schaeumt,1)\n" + "(himmelslichts,1)\n" + "(unbedingte,1)\n" + "(herzlich,1)\n"
+ + "(anblick,1)\n" + "(anblick,2)\n" + "(nennt,1)\n" + "(gruent,1)\n" + "(bluet,1)\n" + "(leichnam,1)\n" + "(erschlaffen,1)\n"
+ + "(jammertagen,1)\n" + "(zieh,1)\n" + "(ihm,1)\n" + "(ihm,2)\n" + "(ihm,3)\n" + "(besondre,1)\n" + "(ihn,1)\n" + "(ihn,2)\n"
+ + "(ihn,3)\n" + "(ihn,4)\n" + "(ihn,5)\n" + "(grossen,1)\n" + "(vollen,1)\n" + "(ihr,1)\n" + "(ihr,2)\n" + "(ihr,3)\n"
+ + "(ihr,4)\n" + "(ihr,5)\n" + "(ihr,6)\n" + "(ihr,7)\n" + "(boten,1)\n" + "(voller,1)\n" + "(singt,1)\n" + "(muhme,1)\n"
+ + "(schon,1)\n" + "(last,1)\n" + "(kleine,1)\n" + "(paradieseshelle,1)\n" + "(nein,1)\n" + "(echten,1)\n" + "(unter,1)\n"
+ + "(bei,1)\n" + "(herr,1)\n" + "(herr,2)\n" + "(herr,3)\n" + "(herr,4)\n" + "(herr,5)\n" + "(herr,6)\n" + "(herr,7)\n"
+ + "(herr,8)\n" + "(herr,9)\n" + "(herr,10)\n" + "(herr,11)\n" + "(gern,1)\n" + "(gern,2)\n" + "(gern,3)\n" + "(sphaerenlauf,1)\n"
+ + "(stets,1)\n" + "(ganze,1)\n" + "(braechte,1)\n" + "(fordert,1)\n" + "(schoensten,1)\n" + "(herrlich,1)\n" + "(herrlich,2)\n"
+ + "(gegeben,1)\n" + "(allein,1)\n" + "(allein,2)\n" + "(reichen,1)\n" + "(schauervoller,1)\n" + "(musst,1)\n" + "(recht,1)\n"
+ + "(bleibt,1)\n" + "(pracht,1)\n" + "(treibt,1)\n" + "(befangen,1)\n" + "(was,1)\n" + "(was,2)\n" + "(menschen,1)\n"
+ + "(menschen,2)\n" + "(menschen,3)\n" + "(jede,1)\n" + "(hohe,1)\n" + "(tiefsten,1)\n" + "(bilden,1)\n" + "(drum,1)\n"
+ + "(gibt,1)\n" + "(gibt,2)\n" + "(guter,1)\n" + "(fuerwahr,1)\n" + "(im,1)\n" + "(im,2)\n" + "(im,3)\n" + "(grund,1)\n"
+ + "(in,1)\n" + "(in,2)\n" + "(in,3)\n" + "(in,4)\n" + "(in,5)\n" + "(in,6)\n" + "(in,7)\n" + "(in,8)\n" + "(in,9)\n" + "(hoechste,1)\n"
+ + "(schliesst,1)\n" + "(fels,1)\n" + "(steh,1)\n" + "(euer,1)\n" + "(erster,1)\n" + "(ersten,1)\n" + "(ersten,2)\n" + "(ersten,3)\n"
+ + "(goettersoehne,1)\n" + "(brechen,1)\n" + "(tiefen,1)\n" + "(frucht,1)\n" + "(kreis,1)\n" + "(siehst,1)\n" + "(wege,1)\n"
+ + "(ist,1)\n" + "(ist,2)\n" + "(ist,3)\n" + "(ist,4)\n" + "(ist,5)\n" + "(ist,6)\n" + "(ist,7)\n" + "(ist,8)\n" + "(zikaden,1)\n"
+ + "(frischen,1)\n" + "(ruh,1)\n" + "(deine,1)\n" + "(deine,2)\n" + "(maus,1)\n" + "(brudersphaeren,1)\n" + "(nachher,1)\n"
+ + "(euch,1)\n" + "(euch,2)\n" + "(euch,3)\n" + "(euch,4)\n" + "(gnaden,1)\n" + "(anzuklagen,1)\n" + "(schlange,1)\n" + "(staerke,1)\n"
+ + "(staerke,2)\n" + "(erde,1)\n" + "(erde,2)\n" + "(erde,3)\n" + "(erde,4)\n" + "(verlaub,1)\n" + "(sanfte,1)\n" + "(holden,1)\n"
+ + "(sonst,1)\n" + "(treten,1)\n" + "(sahst,1)\n" + "(alten,1)\n" + "(um,1)\n" + "(wieder,1)\n" + "(alter,1)\n" + "(altes,1)\n"
+ + "(nun,1)\n" + "(lieb,1)\n" + "(gesellen,1)\n" + "(erscheinen,1)\n" + "(wirkt,1)\n" + "(wirkt,2)\n" + "(haettst,1)\n" + "(haettst,2)\n"
+ + "(nur,1)\n" + "(nur,2)\n" + "(nur,3)\n" + "(nur,4)\n" + "(nur,5)\n" + "(nur,6)\n" + "(nur,7)\n" + "(tiefbewegte,1)\n" + "(lachen,1)\n"
+ + "(lachen,2)\n" + "(drange,1)\n" + "(schlag,1)\n" + "(schein,1)\n" + "(muss,1)\n" + "(verworren,1)\n" + "(weges,1)\n" + "(allen,1)\n"
+ + "(gewoehnlich,1)\n" + "(alles,1)\n" + "(halb,1)\n" + "(stuerme,1)\n" + "(springt,1)\n" + "(sollt,1)\n" + "(klarheit,1)\n"
+ + "(so,1)\n" + "(so,2)\n" + "(so,3)\n" + "(so,4)\n" + "(so,5)\n" + "(so,6)\n" + "(erfassen,1)\n" + "(liedchen,1)\n" + "(prolog,1)\n"
+ + "(zur,1)\n" + "(fressen,1)\n" + "(zum,1)\n" + "(faust,1)\n" + "(faust,2)\n" + "(erzengel,1)\n" + "(erzengel,2)\n" + "(jahre,1)\n"
+ + "(sonn,1)\n" + "(raphael,1)\n" + "(land,1)\n" + "(land,2)\n" + "(lang,1)\n" + "(gelange,1)\n" + "(lust,1)\n" + "(lust,2)\n"
+ + "(welt,1)\n" + "(sehe,1)\n" + "(ihre,1)\n" + "(jedes,1)\n" + "(erfreut,1)\n" + "(seiner,1)\n" + "(denn,1)\n" + "(wandeln,1)\n"
+ + "(wechselt,1)\n" + "(jeden,1)\n" + "(dort,1)\n" + "(schlecht,1)\n" + "(wenigsten,1)\n" + "(wuerd,1)\n" + "(schranken,1)\n"
+ + "(bewusst,1)\n" + "(bewusst,2)\n" + "(seinem,1)\n" + "(seinem,2)\n" + "(gehasst,1)\n" + "(sein,1)\n" + "(meinem,1)\n"
+ + "(meinen,1)\n" + "(pathos,1)\n" + "(herrn,1)\n" + "(lange,1)\n" + "(lange,2)\n" + "(herab,1)\n" + "(diesen,1)\n" + "(ihren,1)\n"
+ + "(beruehmte,1)\n" + "(goethe,1)\n" + "(tag,1)\n" + "(tag,2)\n" + "(tag,3)\n" + "(tier,1)\n" + "(quark,1)\n" + "(dank,1)\n"
+ + "(seine,1)\n" + "(teufel,1)\n" + "(teufel,2)\n" + "(zweck,1)\n" + "(wenn,1)\n" + "(wenn,2)\n" + "(wenn,3)\n" + "(wenn,4)\n"
+ + "(wenn,5)\n" + "(wenn,6)\n" + "(wenn,7)\n" + "(soll,1)\n" + "(wirkung,1)\n" + "(erlaubnis,1)\n" + "(lebendig,1)\n" + "(uns,1)\n"
+ + "(leicht,1)\n" + "(gewiss,1)\n" + "(schnell,1)\n" + "(und,1)\n" + "(und,2)\n" + "(und,3)\n" + "(und,4)\n" + "(und,5)\n" + "(und,6)\n"
+ + "(und,7)\n" + "(und,8)\n" + "(und,9)\n" + "(und,10)\n" + "(und,11)\n" + "(und,12)\n" + "(und,13)\n" + "(und,14)\n" + "(und,15)\n"
+ + "(und,16)\n" + "(und,17)\n" + "(und,18)\n" + "(und,19)\n" + "(und,20)\n" + "(und,21)\n" + "(und,22)\n" + "(und,23)\n" + "(und,24)\n"
+ + "(und,25)\n" + "(und,26)\n" + "(und,27)\n" + "(und,28)\n" + "(und,29)\n" + "(gerne,1)\n" + "(rechten,1)\n" + "(umher,1)\n" + "(umher,2)\n"
+ + "(vernunft,1)\n" + "(grase,1)\n" + "(nach,1)\n" + "(leben,1)\n" + "(gott,1)\n" + "(der,1)\n" + "(der,2)\n" + "(der,3)\n" + "(der,4)\n"
+ + "(der,5)\n" + "(der,6)\n" + "(der,7)\n" + "(der,8)\n" + "(der,9)\n" + "(der,10)\n" + "(der,11)\n" + "(der,12)\n" + "(der,13)\n"
+ + "(der,14)\n" + "(der,15)\n" + "(der,16)\n" + "(der,17)\n" + "(der,18)\n" + "(der,19)\n" + "(der,20)\n" + "(der,21)\n" + "(der,22)\n"
+ + "(der,23)\n" + "(der,24)\n" + "(der,25)\n" + "(der,26)\n" + "(der,27)\n" + "(der,28)\n" + "(der,29)\n" + "(des,1)\n" + "(des,2)\n"
+ + "(des,3)\n" + "(des,4)\n" + "(des,5)\n" + "(doktor,1)\n" + "(beschaemt,1)\n" + "(dreht,1)\n" + "(habe,1)\n" + "(sagen,1)\n" + "(sagen,2)\n"
+ + "(bekennen,1)\n" + "(dunklen,1)\n" + "(wettet,1)\n" + "(den,1)\n" + "(den,2)\n" + "(den,3)\n" + "(den,4)\n" + "(den,5)\n" + "(den,6)\n"
+ + "(den,7)\n" + "(den,8)\n" + "(den,9)\n" + "(mephistopheles,1)\n" + "(mephistopheles,2)\n" + "(mephistopheles,3)\n"
+ + "(mephistopheles,4)\n" + "(mephistopheles,5)\n" + "(mephistopheles,6)\n" + "(mephistopheles,7)\n" + "(mephistopheles,8)\n"
+ + "(mephistopheles,9)\n" + "(dem,1)\n" + "(dem,2)\n" + "(dem,3)\n" + "(dem,4)\n" + "(auch,1)\n" + "(auch,2)\n" + "(auch,3)\n" + "(auch,4)\n"
+ + "(kann,1)\n" + "(kann,2)\n" + "(armen,1)\n" + "(mir,1)\n" + "(mir,2)\n" + "(mir,3)\n" + "(mir,4)\n" + "(mir,5)\n" + "(mir,6)\n" + "(mir,7)\n"
+ + "(mir,8)\n" + "(mir,9)\n" + "(strebt,1)\n" + "(gut,1)\n" + "(gut,2)\n" + "(mit,1)\n" + "(mit,2)\n" + "(mit,3)\n" + "(mit,4)\n" + "(mit,5)\n"
+ + "(mit,6)\n" + "(mit,7)\n" + "(mit,8)\n" + "(mit,9)\n" + "(mit,10)\n" + "(mit,11)\n" + "(bald,1)\n" + "(bald,2)\n" + "(himmlischen,1)\n"
+ + "(himmel,1)\n" + "(himmel,2)\n" + "(himmel,3)\n" + "(noch,1)\n" + "(noch,2)\n" + "(noch,3)\n" + "(kannst,1)\n" + "(deinesgleichen,1)\n"
+ + "(flammt,1)\n" + "(ergruenden,1)\n" + "(ergruenden,2)\n" + "(nacht,1)\n" + "(scheint,1)\n" + "(ferne,1)\n" + "(ferne,2)\n"
+ + "(tragoedie,1)\n" + "(abgewoehnt,1)\n" + "(reizt,1)\n" + "(geistern,1)\n" + "(nicht,1)\n" + "(nicht,2)\n" + "(nicht,3)\n" + "(nicht,4)\n"
+ + "(nicht,5)\n" + "(nicht,6)\n" + "(nicht,7)\n" + "(nicht,8)\n" + "(nicht,9)\n" + "(nicht,10)\n" + "(sacht,1)\n" + "(unbegreiflich,1)\n"
+ + "(unbegreiflich,2)\n" + "(schnelle,1)\n" + "(einmal,1)\n" + "(werd,1)\n" + "(werke,1)\n" + "(werke,2)\n" + "(begraebt,1)\n"
+ + "(knecht,1)\n" + "(rings,1)\n" + "(wird,1)\n" + "(katze,1)\n" + "(huete,1)\n" + "(fortgerissen,1)\n" + "(gebt,1)\n" + "(huebsch,1)\n"
+ + "(hast,1)\n" + "(irrt,1)\n" + "(befinde,1)\n" + "(sind,1)\n" + "(sind,2)\n" + "(fuehren,1)\n" + "(fuehren,2)\n" + "(fliegt,1)\n"
+ + "(ewig,1)\n" + "(ewig,2)\n" + "(ewig,3)\n" + "(brust,1)\n" + "(brust,2)\n" + "(sonne,1)\n" + "(sprechen,1)\n" + "(ein,1)\n" + "(ein,2)\n"
+ + "(ein,3)\n" + "(strasse,1)\n" + "(von,1)\n" + "(von,2)\n" + "(von,3)\n" + "(von,4)\n" + "(von,5)\n" + "(von,6)\n" + "(von,7)\n" + "(von,8)\n"
+ + "(ueberlassen,1)\n" + "(dir,1)\n" + "(dir,2)\n" + "(dir,3)\n" + "(dir,4)\n" + "(vom,1)\n" + "(vom,2)\n" + "(vom,3)\n" + "(zu,1)\n" + "(zu,2)\n"
+ + "(zu,3)\n" + "(zu,4)\n" + "(zu,5)\n" + "(zu,6)\n" + "(zu,7)\n" + "(zu,8)\n" + "(zu,9)\n" + "(zu,10)\n" + "(zu,11)\n" + "(schwebt,1)\n"
+ + "(die,1)\n" + "(die,2)\n" + "(die,3)\n" + "(die,4)\n" + "(die,5)\n" + "(die,6)\n" + "(die,7)\n" + "(die,8)\n" + "(die,9)\n" + "(die,10)\n"
+ + "(die,11)\n" + "(die,12)\n" + "(die,13)\n" + "(die,14)\n" + "(die,15)\n" + "(die,16)\n" + "(die,17)\n" + "(die,18)\n" + "(die,19)\n"
+ + "(die,20)\n" + "(die,21)\n" + "(die,22)\n" + "(vor,1)\n" + "(vor,2)\n" + "(wangen,1)\n" + "(wettgesang,1)\n" + "(donnerschlags,1)\n"
+ + "(find,1)\n" + "(dich,1)\n" + "(dich,2)\n" + "(dich,3)\n" + "(umfass,1)\n" + "(verboten,1)\n" + "(laeg,1)\n" + "(nie,1)\n" + "(drei,1)\n"
+ + "(drei,2)\n" + "(dauern,1)\n" + "(toren,1)\n" + "(dauert,1)\n" + "(verheeren,1)\n" + "(fliegend,1)\n" + "(aus,1)\n" + "(staub,1)\n"
+ + "(fluessen,1)\n" + "(haus,1)\n" + "(auf,1)\n" + "(auf,2)\n" + "(auf,3)\n" + "(auf,4)\n" + "(auf,5)\n" + "(dient,1)\n" + "(dient,2)\n"
+ + "(tiefer,1)\n" + "(naeh,1)\n" + "(zieren,1)\n";
+
public static final String COUNTS_AS_TUPLES = "(machen,1)\n" + "(zeit,2)\n" + "(heerscharen,1)\n" + "(keiner,2)\n" + "(meine,3)\n"
+ "(fuehr,1)\n" + "(triumph,1)\n" + "(kommst,1)\n" + "(frei,1)\n" + "(schaffen,1)\n" + "(gesinde,1)\n"
+ "(langbeinigen,1)\n" + "(schalk,1)\n" + "(besser,1)\n" + "(solang,1)\n" + "(meer,4)\n" + "(fragst,1)\n"
@@ -224,6 +323,7 @@ public class WordCountData {
+ "(verboten,1)\n" + "(laeg,1)\n" + "(nie,1)\n" + "(drei,2)\n" + "(dauern,1)\n" + "(toren,1)\n" + "(dauert,1)\n"
+ "(verheeren,1)\n" + "(fliegend,1)\n" + "(aus,1)\n" + "(staub,1)\n" + "(fluessen,1)\n" + "(haus,1)\n" + "(auf,5)\n"
+ "(dient,2)\n" + "(tiefer,1)\n" + "(naeh,1)\n" + "(zieren,1)\n";
-
- private WordCountData() {}
+
+ private WordCountData() {
+ }
}
[2/2] flink git commit: [FLINK-1560] [streaming]
StreamExecutionEnvironment enhancement for ITCases
Posted by mb...@apache.org.
[FLINK-1560] [streaming] StreamExecutionEnvironment enhancement for ITCases
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6049398d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6049398d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6049398d
Branch: refs/heads/master
Commit: 6049398d5ce7979b19b640124240855e4e8538ef
Parents: c1a334e
Author: mbalassi <mb...@apache.org>
Authored: Wed Feb 18 16:19:13 2015 +0100
Committer: mbalassi <mb...@apache.org>
Committed: Thu Feb 19 10:40:34 2015 +0100
----------------------------------------------------------------------
.../environment/StreamExecutionEnvironment.java | 7 ++
.../StreamExecutionEnvironmentFactory.java | 31 +++++
.../util/StreamingProgramTestBase.java | 118 +++++++++++++++++++
.../streaming/util/TestStreamEnvironment.java | 53 ++++++---
flink-tests/pom.xml | 3 +-
5 files changed, 197 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6049398d/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index de45446..4a3e5a2 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -73,6 +73,8 @@ public abstract class StreamExecutionEnvironment {
protected StreamGraph streamGraph;
+ private static StreamExecutionEnvironmentFactory executionEnvironmentFactory;
+
// --------------------------------------------------------------------------------------------
// Constructor and Properties
// --------------------------------------------------------------------------------------------
@@ -683,4 +685,9 @@ public abstract class StreamExecutionEnvironment {
return getStreamGraph().getStreamingPlanAsJSON();
}
+
+ protected static void initializeFromFactory(StreamExecutionEnvironmentFactory eef) {
+ currentEnvironment = eef.createExecutionEnvironment();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6049398d/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentFactory.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentFactory.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentFactory.java
new file mode 100644
index 0000000..0e24bac
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentFactory.java
@@ -0,0 +1,31 @@
+/*
+ * 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.environment;
+
+/**
+ * Factory class for stream execution environments.
+ */
+public interface StreamExecutionEnvironmentFactory {
+
+ /**
+ * Creates a StreamExecutionEnvironment from this factory.
+ *
+ * @return A StreamExecutionEnvironment.
+ */
+ StreamExecutionEnvironment createExecutionEnvironment();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6049398d/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/StreamingProgramTestBase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/StreamingProgramTestBase.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/StreamingProgramTestBase.java
new file mode 100644
index 0000000..d853846
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/StreamingProgramTestBase.java
@@ -0,0 +1,118 @@
+/*
+ * 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.util;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.junit.Assert;
+
+import org.junit.Test;
+
+public abstract class StreamingProgramTestBase extends AbstractTestBase {
+
+ private static final int DEFAULT_DEGREE_OF_PARALLELISM = 4;
+
+ private JobExecutionResult latestExecutionResult;
+
+ private int degreeOfParallelism = DEFAULT_DEGREE_OF_PARALLELISM;
+
+
+ public StreamingProgramTestBase() {
+ this(new Configuration());
+ }
+
+ public StreamingProgramTestBase(Configuration config) {
+ super(config);
+ setTaskManagerNumSlots(degreeOfParallelism);
+ }
+
+ public void setDegreeOfParallelism(int degreeOfParallelism) {
+ this.degreeOfParallelism = degreeOfParallelism;
+ setTaskManagerNumSlots(degreeOfParallelism);
+ }
+
+ public int getDegreeOfParallelism() {
+ return degreeOfParallelism;
+ }
+
+ public JobExecutionResult getLatestExecutionResult() {
+ return this.latestExecutionResult;
+ }
+
+
+ // --------------------------------------------------------------------------------------------
+ // Methods to create the test program and for pre- and post- test work
+ // --------------------------------------------------------------------------------------------
+
+ protected abstract void testProgram() throws Exception;
+
+ protected void preSubmit() throws Exception {}
+
+ protected void postSubmit() throws Exception {}
+
+ // --------------------------------------------------------------------------------------------
+ // Test entry point
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testJobWithoutObjectReuse() throws Exception {
+ startCluster();
+ try {
+ // pre-submit
+ try {
+ preSubmit();
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail("Pre-submit work caused an error: " + e.getMessage());
+ }
+
+ // prepare the test environment
+ TestStreamEnvironment env = new TestStreamEnvironment(this.executor, this.degreeOfParallelism);
+ env.setAsContext();
+
+ // call the test program
+ try {
+ testProgram();
+ this.latestExecutionResult = env.latestResult;
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail("Error while calling the test program: " + e.getMessage());
+ }
+
+ Assert.assertNotNull("The test program never triggered an execution.", this.latestExecutionResult);
+
+ // post-submit
+ try {
+ postSubmit();
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail("Post-submit work caused an error: " + e.getMessage());
+ }
+ } finally {
+ stopCluster();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6049398d/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestStreamEnvironment.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
index e1b3ac0..6e0821d 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
@@ -19,12 +19,14 @@
package org.apache.flink.streaming.util;
import akka.actor.ActorRef;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.client.JobClient;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory;
import org.apache.flink.test.util.ForkableFlinkMiniCluster;
public class TestStreamEnvironment extends StreamExecutionEnvironment {
@@ -32,11 +34,19 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment {
private static final String CANNOT_EXECUTE_EMPTY_JOB = "Cannot execute empty job";
private long memorySize;
+ protected JobExecutionResult latestResult;
+ private ForkableFlinkMiniCluster executor;
+ private boolean internalExecutor;
public TestStreamEnvironment(int degreeOfParallelism, long memorySize){
- this.setDegreeOfParallelism(degreeOfParallelism);
-
+ setDegreeOfParallelism(degreeOfParallelism);
this.memorySize = memorySize;
+ internalExecutor = true;
+ }
+
+ public TestStreamEnvironment(ForkableFlinkMiniCluster executor, int dop){
+ this.executor = executor;
+ setDefaultLocalParallelism(dop);
}
@Override
@@ -48,25 +58,40 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment {
public void execute(String jobName) throws Exception {
JobGraph jobGraph = streamGraph.getJobGraph(jobName);
- Configuration configuration = jobGraph.getJobConfiguration();
+ if (internalExecutor) {
+ Configuration configuration = jobGraph.getJobConfiguration();
- configuration.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS,
- getDegreeOfParallelism());
- configuration.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, memorySize);
-
- ForkableFlinkMiniCluster cluster = new ForkableFlinkMiniCluster(configuration);
+ configuration.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS,
+ getDegreeOfParallelism());
+ configuration.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, memorySize);
+ executor = new ForkableFlinkMiniCluster(configuration);
+ }
try {
- ActorRef client = cluster.getJobClient();
- JobClient.submitJobAndWait(jobGraph, false, client, cluster.timeout());
- } catch(JobExecutionException e){
- if(e.getMessage().contains("GraphConversionException")){
+ ActorRef client = executor.getJobClient();
+ latestResult = JobClient.submitJobAndWait(jobGraph, false, client, executor.timeout());
+ } catch(JobExecutionException e) {
+ if (e.getMessage().contains("GraphConversionException")) {
throw new Exception(CANNOT_EXECUTE_EMPTY_JOB, e);
} else {
throw e;
}
- } finally{
- cluster.stop();
+ } finally {
+ if (internalExecutor){
+ executor.shutdown();
+ }
}
}
+
+ protected void setAsContext() {
+ StreamExecutionEnvironmentFactory factory = new StreamExecutionEnvironmentFactory() {
+ @Override
+ public StreamExecutionEnvironment createExecutionEnvironment() {
+ return TestStreamEnvironment.this;
+ }
+ };
+
+ initializeFromFactory(factory);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6049398d/flink-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-tests/pom.xml b/flink-tests/pom.xml
index e1db825..17c6b3e 100644
--- a/flink-tests/pom.xml
+++ b/flink-tests/pom.xml
@@ -85,7 +85,7 @@ under the License.
<version>${project.version}</version>
<scope>test</scope>
</dependency>
-
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils</artifactId>
@@ -106,6 +106,7 @@ under the License.
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>