You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2014/12/18 19:46:12 UTC
[76/82] [abbrv] incubator-flink git commit: Change integration tests
to reuse cluster in order to save startup and shutdown time.
Change integration tests to reuse cluster in order to save startup and shutdown time.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/a0f77857
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/a0f77857
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/a0f77857
Branch: refs/heads/master
Commit: a0f7785720f786a1cf3a70ed31b9d05cfd57c65b
Parents: f10d357
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Dec 12 15:55:48 2014 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Dec 18 18:58:32 2014 +0100
----------------------------------------------------------------------
.../mapred/HadoopMapFunctionITCase.java | 181 +--
.../HadoopReduceCombineFunctionITCase.java | 340 +++--
.../mapred/HadoopReduceFunctionITCase.java | 231 ++--
.../minicluster/LocalFlinkMiniCluster.scala | 8 +-
.../flink/test/util/AbstractTestBase.java | 348 +----
.../test/util/CollectionTestEnvironment.java | 52 +
.../flink/test/util/JavaProgramTestBase.java | 103 --
.../test/util/MultipleProgramsTestBase.java | 66 +
.../apache/flink/test/util/TestBaseUtils.java | 353 +++++
.../apache/flink/test/util/TestEnvironment.java | 96 ++
.../test/util/ForkableFlinkMiniCluster.scala | 9 +-
.../exampleJavaPrograms/PageRankITCase.java | 100 +-
.../exampleScalaPrograms/PageRankITCase.java | 108 +-
.../aggregators/AggregatorsITCase.java | 362 +++--
.../test/javaApiOperators/AggregateITCase.java | 178 +--
.../test/javaApiOperators/CoGroupITCase.java | 913 ++++++------
.../test/javaApiOperators/CrossITCase.java | 640 ++++-----
.../test/javaApiOperators/DataSourceITCase.java | 80 +-
.../test/javaApiOperators/DistinctITCase.java | 501 ++++---
.../test/javaApiOperators/FilterITCase.java | 577 ++++----
.../test/javaApiOperators/FirstNITCase.java | 181 +--
.../test/javaApiOperators/FlatMapITCase.java | 654 +++++----
.../javaApiOperators/GroupReduceITCase.java | 1269 ++++++++---------
.../flink/test/javaApiOperators/JoinITCase.java | 1240 ++++++++---------
.../flink/test/javaApiOperators/MapITCase.java | 881 ++++++------
.../test/javaApiOperators/PartitionITCase.java | 385 +++---
.../test/javaApiOperators/ProjectITCase.java | 108 +-
.../test/javaApiOperators/ReduceITCase.java | 684 +++++----
.../test/javaApiOperators/SumMinMaxITCase.java | 172 +--
.../test/javaApiOperators/UnionITCase.java | 234 ++--
.../scala/functions/ClosureCleanerITCase.scala | 136 +-
.../api/scala/operators/AggregateITCase.scala | 152 +-
.../api/scala/operators/CoGroupITCase.scala | 643 +++++----
.../flink/api/scala/operators/CrossITCase.scala | 378 +++--
.../api/scala/operators/DistinctITCase.scala | 288 ++--
.../api/scala/operators/ExamplesITCase.scala | 176 ++-
.../api/scala/operators/FilterITCase.scala | 245 ++--
.../api/scala/operators/FirstNITCase.scala | 133 +-
.../api/scala/operators/FlatMapITCase.scala | 328 +++--
.../api/scala/operators/GroupReduceITCase.scala | 1302 +++++++++---------
.../flink/api/scala/operators/JoinITCase.scala | 641 ++++-----
.../flink/api/scala/operators/MapITCase.scala | 382 +++--
.../api/scala/operators/PartitionITCase.scala | 320 ++---
.../api/scala/operators/ReduceITCase.scala | 376 +++--
.../api/scala/operators/SumMinMaxITCase.scala | 158 +--
.../flink/api/scala/operators/UnionITCase.scala | 139 +-
.../scala/runtime/ScalaSpecialTypesITCase.scala | 215 ++-
47 files changed, 8283 insertions(+), 8783 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopMapFunctionITCase.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopMapFunctionITCase.java b/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopMapFunctionITCase.java
index 6a9ad2c..850e799 100644
--- a/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopMapFunctionITCase.java
+++ b/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopMapFunctionITCase.java
@@ -18,154 +18,115 @@
package org.apache.flink.test.hadoopcompatibility.mapred;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.util.Collection;
-import java.util.LinkedList;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.hadoopcompatibility.mapred.HadoopMapFunction;
-import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class HadoopMapFunctionITCase extends JavaProgramTestBase {
+public class HadoopMapFunctionITCase extends MultipleProgramsTestBase {
- private static int NUM_PROGRAMS = 3;
-
- private int curProgId = config.getInteger("ProgramId", -1);
- private String resultPath;
- private String expectedResult;
-
- public HadoopMapFunctionITCase(Configuration config) {
- super(config);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ public HadoopMapFunctionITCase(ExecutionMode mode){
+ super(mode);
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = MapperProgs.runProgram(curProgId, resultPath);
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
- }
-
- @Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @Test
+ public void testNonPassingMapper() throws Exception{
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
-
- return toParameterList(tConfigs);
+ DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
+ DataSet<Tuple2<IntWritable, Text>> nonPassingFlatMapDs = ds.
+ flatMap(new HadoopMapFunction<IntWritable, Text, IntWritable, Text>(new NonPassingMapper()));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ nonPassingFlatMapDs.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+
+ compareResultsByLinesInMemory("\n", resultPath);
}
-
- public static class MapperProgs {
-
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch(progId) {
- case 1: {
- /*
- * Test non-passing mapper
- */
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
- DataSet<Tuple2<IntWritable, Text>> nonPassingFlatMapDs = ds.
- flatMap(new HadoopMapFunction<IntWritable, Text, IntWritable, Text>(new NonPassingMapper()));
-
- nonPassingFlatMapDs.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "\n";
- }
- case 2: {
- /*
- * Test data duplicating mapper
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
- DataSet<Tuple2<IntWritable, Text>> duplicatingFlatMapDs = ds.
- flatMap(new HadoopMapFunction<IntWritable, Text, IntWritable, Text>(new DuplicatingMapper()));
-
- duplicatingFlatMapDs.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(1,Hi)\n" + "(1,HI)\n" +
+
+ @Test
+ public void testDataDuplicatingMapper() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
+ DataSet<Tuple2<IntWritable, Text>> duplicatingFlatMapDs = ds.
+ flatMap(new HadoopMapFunction<IntWritable, Text, IntWritable, Text>(new DuplicatingMapper()));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ duplicatingFlatMapDs.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+
+ String expected = "(1,Hi)\n" + "(1,HI)\n" +
"(2,Hello)\n" + "(2,HELLO)\n" +
"(3,Hello world)\n" + "(3,HELLO WORLD)\n" +
"(4,Hello world, how are you?)\n" + "(4,HELLO WORLD, HOW ARE YOU?)\n" +
"(5,I am fine.)\n" + "(5,I AM FINE.)\n" +
"(6,Luke Skywalker)\n" + "(6,LUKE SKYWALKER)\n" +
- "(7,Comment#1)\n" + "(7,COMMENT#1)\n" +
- "(8,Comment#2)\n" + "(8,COMMENT#2)\n" +
- "(9,Comment#3)\n" + "(9,COMMENT#3)\n" +
- "(10,Comment#4)\n" + "(10,COMMENT#4)\n" +
- "(11,Comment#5)\n" + "(11,COMMENT#5)\n" +
- "(12,Comment#6)\n" + "(12,COMMENT#6)\n" +
- "(13,Comment#7)\n" + "(13,COMMENT#7)\n" +
+ "(7,Comment#1)\n" + "(7,COMMENT#1)\n" +
+ "(8,Comment#2)\n" + "(8,COMMENT#2)\n" +
+ "(9,Comment#3)\n" + "(9,COMMENT#3)\n" +
+ "(10,Comment#4)\n" + "(10,COMMENT#4)\n" +
+ "(11,Comment#5)\n" + "(11,COMMENT#5)\n" +
+ "(12,Comment#6)\n" + "(12,COMMENT#6)\n" +
+ "(13,Comment#7)\n" + "(13,COMMENT#7)\n" +
"(14,Comment#8)\n" + "(14,COMMENT#8)\n" +
- "(15,Comment#9)\n" + "(15,COMMENT#9)\n" +
+ "(15,Comment#9)\n" + "(15,COMMENT#9)\n" +
"(16,Comment#10)\n" + "(16,COMMENT#10)\n" +
"(17,Comment#11)\n" + "(17,COMMENT#11)\n" +
"(18,Comment#12)\n" + "(18,COMMENT#12)\n" +
"(19,Comment#13)\n" + "(19,COMMENT#13)\n" +
"(20,Comment#14)\n" + "(20,COMMENT#14)\n" +
"(21,Comment#15)\n" + "(21,COMMENT#15)\n";
- }
- case 3: {
- // Mapper configured via JobConf
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- JobConf conf = new JobConf();
- conf.set("my.filterPrefix", "Hello");
-
- DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
- DataSet<Tuple2<IntWritable, Text>> hellos = ds.
- flatMap(new HadoopMapFunction<IntWritable, Text, IntWritable, Text>(new ConfigurableMapper(), conf));
-
- hellos.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(2,Hello)\n" +
+
+ compareResultsByLinesInMemory(expected, resultPath);
+ }
+
+ @Test
+ public void testConfigurableMapper() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ JobConf conf = new JobConf();
+ conf.set("my.filterPrefix", "Hello");
+
+ DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
+ DataSet<Tuple2<IntWritable, Text>> hellos = ds.
+ flatMap(new HadoopMapFunction<IntWritable, Text, IntWritable, Text>(new ConfigurableMapper(), conf));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ hellos.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+
+ String expected = "(2,Hello)\n" +
"(3,Hello world)\n" +
"(4,Hello world, how are you?)\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
-
- }
-
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
+
+
public static class NonPassingMapper implements Mapper<IntWritable, Text, IntWritable, Text> {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceCombineFunctionITCase.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceCombineFunctionITCase.java b/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceCombineFunctionITCase.java
index e2c124d..92b0dc3 100644
--- a/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceCombineFunctionITCase.java
+++ b/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceCombineFunctionITCase.java
@@ -18,219 +18,139 @@
package org.apache.flink.test.hadoopcompatibility.mapred;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.util.Collection;
import java.util.Iterator;
-import java.util.LinkedList;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.hadoopcompatibility.mapred.HadoopReduceCombineFunction;
import org.apache.flink.hadoopcompatibility.mapred.HadoopReduceFunction;
-import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
+import org.hamcrest.core.IsEqual;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class HadoopReduceCombineFunctionITCase extends JavaProgramTestBase {
+public class HadoopReduceCombineFunctionITCase extends MultipleProgramsTestBase {
- private static int NUM_PROGRAMS = 4;
-
- private int curProgId = config.getInteger("ProgramId", -1);
- private String resultPath;
- private String expectedResult;
-
- public HadoopReduceCombineFunctionITCase(Configuration config) {
- super(config);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ public HadoopReduceCombineFunctionITCase(ExecutionMode mode){
+ super(mode);
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = ReducerProgs.runProgram(curProgId, resultPath);
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Test
+ public void testStandardCountingWithCombiner() throws Exception{
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Tuple2<IntWritable, IntWritable>> ds = HadoopTestData.getKVPairDataSet(env).
+ map(new Mapper1());
+
+ DataSet<Tuple2<IntWritable, IntWritable>> counts = ds.
+ groupBy(0).
+ reduceGroup(new HadoopReduceCombineFunction<IntWritable, IntWritable, IntWritable, IntWritable>(
+ new SumReducer(), new SumReducer()));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ counts.writeAsText(resultPath);
+ env.execute();
+
+ String expected = "(0,5)\n"+
+ "(1,6)\n" +
+ "(2,6)\n" +
+ "(3,4)\n";
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
-
- @Override
- protected boolean skipCollectionExecution() {
- if (this.curProgId == 3) {
- return true;
- }
- return false;
+
+ @Test
+ public void testUngroupedHadoopReducer() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Tuple2<IntWritable, IntWritable>> ds = HadoopTestData.getKVPairDataSet(env).
+ map(new Mapper2());
+
+ DataSet<Tuple2<IntWritable, IntWritable>> sum = ds.
+ reduceGroup(new HadoopReduceCombineFunction<IntWritable, IntWritable, IntWritable, IntWritable>(
+ new SumReducer(), new SumReducer()));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ sum.writeAsText(resultPath);
+ env.execute();
+
+ String expected = "(0,231)\n";
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
-
- @Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @Test
+ public void testCombiner() throws Exception {
+ org.junit.Assume.assumeThat(mode, new IsEqual<ExecutionMode>(ExecutionMode.CLUSTER));
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
-
- return toParameterList(tConfigs);
+ DataSet<Tuple2<IntWritable, IntWritable>> ds = HadoopTestData.getKVPairDataSet(env).
+ map(new Mapper3());
+
+ DataSet<Tuple2<IntWritable, IntWritable>> counts = ds.
+ groupBy(0).
+ reduceGroup(new HadoopReduceCombineFunction<IntWritable, IntWritable, IntWritable, IntWritable>(
+ new SumReducer(), new KeyChangingReducer()));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ counts.writeAsText(resultPath);
+ env.execute();
+
+ String expected = "(0,5)\n"+
+ "(1,6)\n" +
+ "(2,5)\n" +
+ "(3,5)\n";
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
-
- public static class ReducerProgs {
-
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch(progId) {
- case 1: {
- /*
- * Test standard counting with combiner
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<Tuple2<IntWritable, IntWritable>> ds = HadoopTestData.getKVPairDataSet(env).
- map(new MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, IntWritable>>() {
- private static final long serialVersionUID = 1L;
- Tuple2<IntWritable,IntWritable> outT = new Tuple2<IntWritable,IntWritable>();
- @Override
- public Tuple2<IntWritable, IntWritable> map(Tuple2<IntWritable, Text> v)
- throws Exception {
- outT.f0 = new IntWritable(v.f0.get() / 6);
- outT.f1 = new IntWritable(1);
- return outT;
- }
- });
-
- DataSet<Tuple2<IntWritable, IntWritable>> counts = ds.
- groupBy(0).
- reduceGroup(new HadoopReduceCombineFunction<IntWritable, IntWritable, IntWritable, IntWritable>(
- new SumReducer(), new SumReducer()));
-
- counts.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(0,5)\n"+
- "(1,6)\n" +
- "(2,6)\n" +
- "(3,4)\n";
- }
- case 2: {
- /*
- * Test ungrouped Hadoop reducer
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<Tuple2<IntWritable, IntWritable>> ds = HadoopTestData.getKVPairDataSet(env).
- map(new MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, IntWritable>>() {
- private static final long serialVersionUID = 1L;
- Tuple2<IntWritable,IntWritable> outT = new Tuple2<IntWritable,IntWritable>();
- @Override
- public Tuple2<IntWritable, IntWritable> map(Tuple2<IntWritable, Text> v)
- throws Exception {
- outT.f0 = new IntWritable(0);
- outT.f1 = v.f0;
- return outT;
- }
- });
-
- DataSet<Tuple2<IntWritable, IntWritable>> sum = ds.
- reduceGroup(new HadoopReduceCombineFunction<IntWritable, IntWritable, IntWritable, IntWritable>(
- new SumReducer(), new SumReducer()));
-
- sum.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(0,231)\n";
- }
- case 3: {
- /* Test combiner */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<Tuple2<IntWritable, IntWritable>> ds = HadoopTestData.getKVPairDataSet(env).
- map(new MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, IntWritable>>() {
- private static final long serialVersionUID = 1L;
- Tuple2<IntWritable,IntWritable> outT = new Tuple2<IntWritable,IntWritable>();
- @Override
- public Tuple2<IntWritable, IntWritable> map(Tuple2<IntWritable, Text> v)
- throws Exception {
- outT.f0 = v.f0;
- outT.f1 = new IntWritable(1);
- return outT;
- }
- });
-
- DataSet<Tuple2<IntWritable, IntWritable>> counts = ds.
- groupBy(0).
- reduceGroup(new HadoopReduceCombineFunction<IntWritable, IntWritable, IntWritable, IntWritable>(
- new SumReducer(), new KeyChangingReducer()));
-
- counts.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(0,5)\n"+
- "(1,6)\n" +
- "(2,5)\n" +
- "(3,5)\n";
- }
- case 4: {
- /*
- * Test configuration via JobConf
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- JobConf conf = new JobConf();
- conf.set("my.cntPrefix", "Hello");
-
- DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env).
- map(new MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, Text>>() {
- private static final long serialVersionUID = 1L;
- @Override
- public Tuple2<IntWritable, Text> map(Tuple2<IntWritable, Text> v)
- throws Exception {
- v.f0 = new IntWritable(v.f0.get() % 5);
- return v;
- }
- });
-
- DataSet<Tuple2<IntWritable, IntWritable>> hellos = ds.
- groupBy(0).
- reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(
- new ConfigurableCntReducer(), conf));
-
- hellos.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(0,0)\n"+
- "(1,0)\n" +
- "(2,1)\n" +
- "(3,1)\n" +
- "(4,1)\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
-
- }
-
+
+ @Test
+ public void testConfigurationViaJobConf() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ JobConf conf = new JobConf();
+ conf.set("my.cntPrefix", "Hello");
+
+ DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env).
+ map(new Mapper4());
+
+ DataSet<Tuple2<IntWritable, IntWritable>> hellos = ds.
+ groupBy(0).
+ reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(
+ new ConfigurableCntReducer(), conf));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ hellos.writeAsText(resultPath);
+ env.execute();
+
+ // return expected result
+ String expected = "(0,0)\n"+
+ "(1,0)\n" +
+ "(2,1)\n" +
+ "(3,1)\n" +
+ "(4,1)\n";
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
public static class SumReducer implements Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
@@ -294,4 +214,52 @@ public class HadoopReduceCombineFunctionITCase extends JavaProgramTestBase {
@Override
public void close() throws IOException { }
}
+
+ public static class Mapper1 implements MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable,
+ IntWritable>> {
+ private static final long serialVersionUID = 1L;
+ Tuple2<IntWritable,IntWritable> outT = new Tuple2<IntWritable,IntWritable>();
+ @Override
+ public Tuple2<IntWritable, IntWritable> map(Tuple2<IntWritable, Text> v)
+ throws Exception {
+ outT.f0 = new IntWritable(v.f0.get() / 6);
+ outT.f1 = new IntWritable(1);
+ return outT;
+ }
+ }
+
+ public static class Mapper2 implements MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable,
+ IntWritable>> {
+ private static final long serialVersionUID = 1L;
+ Tuple2<IntWritable,IntWritable> outT = new Tuple2<IntWritable,IntWritable>();
+ @Override
+ public Tuple2<IntWritable, IntWritable> map(Tuple2<IntWritable, Text> v)
+ throws Exception {
+ outT.f0 = new IntWritable(0);
+ outT.f1 = v.f0;
+ return outT;
+ }
+ }
+
+ public static class Mapper3 implements MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, IntWritable>> {
+ private static final long serialVersionUID = 1L;
+ Tuple2<IntWritable,IntWritable> outT = new Tuple2<IntWritable,IntWritable>();
+ @Override
+ public Tuple2<IntWritable, IntWritable> map(Tuple2<IntWritable, Text> v)
+ throws Exception {
+ outT.f0 = v.f0;
+ outT.f1 = new IntWritable(1);
+ return outT;
+ }
+ }
+
+ public static class Mapper4 implements MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, Text>> {
+ private static final long serialVersionUID = 1L;
+ @Override
+ public Tuple2<IntWritable, Text> map(Tuple2<IntWritable, Text> v)
+ throws Exception {
+ v.f0 = new IntWritable(v.f0.get() % 5);
+ return v;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceFunctionITCase.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceFunctionITCase.java b/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceFunctionITCase.java
index 605bf21..1801c3e 100644
--- a/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceFunctionITCase.java
+++ b/flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopReduceFunctionITCase.java
@@ -18,165 +18,108 @@
package org.apache.flink.test.hadoopcompatibility.mapred;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.util.Collection;
import java.util.Iterator;
-import java.util.LinkedList;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.hadoopcompatibility.mapred.HadoopReduceFunction;
-import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class HadoopReduceFunctionITCase extends JavaProgramTestBase {
+public class HadoopReduceFunctionITCase extends MultipleProgramsTestBase {
- private static int NUM_PROGRAMS = 3;
-
- private int curProgId = config.getInteger("ProgramId", -1);
- private String resultPath;
- private String expectedResult;
-
- public HadoopReduceFunctionITCase(Configuration config) {
- super(config);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ public HadoopReduceFunctionITCase(ExecutionMode mode){
+ super(mode);
}
- @Override
- protected void testProgram() throws Exception {
- expectedResult = ReducerProgs.runProgram(curProgId, resultPath);
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(expectedResult, resultPath);
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Test
+ public void testStandardGrouping() throws Exception{
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env).
+ map(new Mapper1());
+
+ DataSet<Tuple2<IntWritable, IntWritable>> commentCnts = ds.
+ groupBy(0).
+ reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(new CommentCntReducer()));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ commentCnts.writeAsText(resultPath);
+ env.execute();
+
+ String expected = "(0,0)\n"+
+ "(1,3)\n" +
+ "(2,5)\n" +
+ "(3,5)\n" +
+ "(4,2)\n";
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
-
- @Parameters
- public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @Test
+ public void testUngroupedHadoopReducer() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
-
- return toParameterList(tConfigs);
+ DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
+
+ DataSet<Tuple2<IntWritable, IntWritable>> commentCnts = ds.
+ reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(new AllCommentCntReducer()));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ commentCnts.writeAsText(resultPath);
+ env.execute();
+
+ String expected = "(42,15)\n";
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
-
- public static class ReducerProgs {
-
- public static String runProgram(int progId, String resultPath) throws Exception {
-
- switch(progId) {
- case 1: {
- /*
- * Test standard grouping
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env).
- map(new MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, Text>>() {
- private static final long serialVersionUID = 1L;
- @Override
- public Tuple2<IntWritable, Text> map(Tuple2<IntWritable, Text> v)
- throws Exception {
- v.f0 = new IntWritable(v.f0.get() / 5);
- return v;
- }
- });
-
- DataSet<Tuple2<IntWritable, IntWritable>> commentCnts = ds.
- groupBy(0).
- reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(new CommentCntReducer()));
-
- commentCnts.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(0,0)\n"+
- "(1,3)\n" +
- "(2,5)\n" +
- "(3,5)\n" +
- "(4,2)\n";
- }
- case 2: {
- /*
- * Test ungrouped Hadoop reducer
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env);
-
- DataSet<Tuple2<IntWritable, IntWritable>> commentCnts = ds.
- reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(new AllCommentCntReducer()));
-
- commentCnts.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(42,15)\n";
- }
- case 3: {
- /*
- * Test configuration via JobConf
- */
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- JobConf conf = new JobConf();
- conf.set("my.cntPrefix", "Hello");
-
- DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env).
- map(new MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, Text>>() {
- private static final long serialVersionUID = 1L;
- @Override
- public Tuple2<IntWritable, Text> map(Tuple2<IntWritable, Text> v)
- throws Exception {
- v.f0 = new IntWritable(v.f0.get() % 5);
- return v;
- }
- });
-
- DataSet<Tuple2<IntWritable, IntWritable>> helloCnts = ds.
- groupBy(0).
- reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(
- new ConfigurableCntReducer(), conf));
-
- helloCnts.writeAsText(resultPath);
- env.execute();
-
- // return expected result
- return "(0,0)\n"+
- "(1,0)\n" +
- "(2,1)\n" +
- "(3,1)\n" +
- "(4,1)\n";
- }
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
-
- }
-
+
+ @Test
+ public void testConfigurationViaJobConf() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ JobConf conf = new JobConf();
+ conf.set("my.cntPrefix", "Hello");
+
+ DataSet<Tuple2<IntWritable, Text>> ds = HadoopTestData.getKVPairDataSet(env).
+ map(new Mapper2());
+
+ DataSet<Tuple2<IntWritable, IntWritable>> helloCnts = ds.
+ groupBy(0).
+ reduceGroup(new HadoopReduceFunction<IntWritable, Text, IntWritable, IntWritable>(
+ new ConfigurableCntReducer(), conf));
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+
+ helloCnts.writeAsText(resultPath);
+ env.execute();
+
+ String expected = "(0,0)\n"+
+ "(1,0)\n" +
+ "(2,1)\n" +
+ "(3,1)\n" +
+ "(4,1)\n";
+
+ compareResultsByLinesInMemory(expected, resultPath);
}
public static class CommentCntReducer implements Reducer<IntWritable, Text, IntWritable, IntWritable> {
@@ -247,4 +190,24 @@ public class HadoopReduceFunctionITCase extends JavaProgramTestBase {
@Override
public void close() throws IOException { }
}
+
+ public static class Mapper1 implements MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, Text>> {
+ private static final long serialVersionUID = 1L;
+ @Override
+ public Tuple2<IntWritable, Text> map(Tuple2<IntWritable, Text> v)
+ throws Exception {
+ v.f0 = new IntWritable(v.f0.get() / 5);
+ return v;
+ }
+ }
+
+ public static class Mapper2 implements MapFunction<Tuple2<IntWritable, Text>, Tuple2<IntWritable, Text>> {
+ private static final long serialVersionUID = 1L;
+ @Override
+ public Tuple2<IntWritable, Text> map(Tuple2<IntWritable, Text> v)
+ throws Exception {
+ v.f0 = new IntWritable(v.f0.get() % 5);
+ return v;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
index fbf5d83..c1169d3 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
@@ -69,7 +69,13 @@ FlinkMiniCluster(userConfiguration){
config.setInteger(ConfigConstants.TASK_MANAGER_DATA_PORT_KEY, dataPort + index)
}
- TaskManager.startActorWithConfiguration(HOSTNAME, config, false)(system)
+ val localExecution = if(numTaskManagers == 1){
+ false
+ }else{
+ false
+ }
+
+ TaskManager.startActorWithConfiguration(HOSTNAME, config, localExecution)(system)
}
def getJobClient(): ActorRef ={
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
index 706f919..38b52ca 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
@@ -18,59 +18,24 @@
package org.apache.flink.test.util;
-import java.io.BufferedInputStream;
-import java.io.BufferedReader;
import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileReader;
-import java.io.FilenameFilter;
import java.io.IOException;
-import java.io.OutputStream;
-import java.io.PrintStream;
-import java.net.URI;
-import java.net.URISyntaxException;
import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import akka.actor.ActorRef;
-import akka.dispatch.Futures;
-import akka.pattern.Patterns;
-import akka.util.Timeout;
-import org.apache.commons.io.FileUtils;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
-import org.apache.hadoop.fs.FileSystem;
-import org.junit.Assert;
import com.google.common.base.Charsets;
import com.google.common.io.Files;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
-public abstract class AbstractTestBase {
- protected static final int MINIMUM_HEAP_SIZE_MB = 192;
-
- protected static final long TASK_MANAGER_MEMORY_SIZE = 80;
+public abstract class AbstractTestBase extends TestBaseUtils {
- protected static final int DEFAULT_TASK_MANAGER_NUM_SLOTS = 1;
-
- protected static final int DEFAULT_NUM_TASK_MANAGERS = 1;
-
- protected static final int DEFAULT_AKKA_ASK_TIMEOUT = 1000;
-
- protected static final String DEFAULT_AKKA_STARTUP_TIMEOUT = "60 s";
protected final Configuration config;
-
+
protected ForkableFlinkMiniCluster executor;
private final List<File> tempFiles;
@@ -82,7 +47,6 @@ public abstract class AbstractTestBase {
private final FiniteDuration timeout;
public AbstractTestBase(Configuration config) {
- verifyJvmOptions();
this.config = config;
this.tempFiles = new ArrayList<File>();
@@ -90,111 +54,72 @@ public abstract class AbstractTestBase {
ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT), TimeUnit.SECONDS);
}
- private void verifyJvmOptions() {
- long heap = Runtime.getRuntime().maxMemory() >> 20;
- Assert.assertTrue("Insufficient java heap space " + heap + "mb - set JVM option: -Xmx" + MINIMUM_HEAP_SIZE_MB
- + "m", heap > MINIMUM_HEAP_SIZE_MB - 50);
- }
// --------------------------------------------------------------------------------------------
// Local Test Cluster Life Cycle
// --------------------------------------------------------------------------------------------
-
- public void startCluster() throws Exception {
- Configuration config = new Configuration();
- config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, true);
- config.setBoolean(ConfigConstants.TASK_MANAGER_MEMORY_LAZY_ALLOCATION_KEY, true);
- config.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, TASK_MANAGER_MEMORY_SIZE);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, taskManagerNumSlots);
- config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers);
- config.setInteger(ConfigConstants.AKKA_ASK_TIMEOUT, DEFAULT_AKKA_ASK_TIMEOUT);
- config.setString(ConfigConstants.AKKA_STARTUP_TIMEOUT, DEFAULT_AKKA_STARTUP_TIMEOUT);
- this.executor = new ForkableFlinkMiniCluster(config);
+
+ public void startCluster(){
+ this.executor = startCluster(numTaskManagers, taskManagerNumSlots);
}
public void stopCluster() throws Exception {
- try {
-
- int numUnreleasedBCVars = 0;
+ stopCluster(executor, timeout);
int numActiveConnections = 0;
- {
- List<ActorRef> tms = executor.getTaskManagersAsJava();
- List<Future<Object>> responseFutures = new ArrayList<Future<Object>>();
-
- for(ActorRef tm: tms){
- responseFutures.add(Patterns.ask(tm, TestingTaskManagerMessages
- .RequestBroadcastVariablesWithReferences$.MODULE$, new Timeout
- (timeout)));
- }
-
- Future<Iterable<Object>> futureResponses = Futures.sequence(
- responseFutures, AkkaUtils.globalExecutionContext());
-
- Iterable<Object> responses = Await.result(futureResponses, timeout);
-
- for(Object response: responses) {
- numUnreleasedBCVars += ((TestingTaskManagerMessages
- .ResponseBroadcastVariablesWithReferences) response).number();
- }
- }
-
- if (this.executor != null) {
- this.executor.stop();
- this.executor = null;
- FileSystem.closeAll();
- System.gc();
- }
-
- Assert.assertEquals("Not all broadcast variables were released.", 0, numUnreleasedBCVars);
Assert.assertEquals("Not all network connections were released.", 0, numActiveConnections);
- }
- finally {
- deleteAllTempFiles();
- }
+ deleteAllTempFiles();
}
//------------------
// Accessors
//------------------
- public int getTaskManagerNumSlots() { return taskManagerNumSlots; }
+ public int getTaskManagerNumSlots() {
+ return taskManagerNumSlots;
+ }
- public void setTaskManagerNumSlots(int taskManagerNumSlots) { this.taskManagerNumSlots = taskManagerNumSlots; }
+ public void setTaskManagerNumSlots(int taskManagerNumSlots) {
+ this.taskManagerNumSlots = taskManagerNumSlots;
+ }
- public int getNumTaskManagers() { return numTaskManagers; }
+ public int getNumTaskManagers() {
+ return numTaskManagers;
+ }
+
+ public void setNumTaskManagers(int numTaskManagers) {
+ this.numTaskManagers = numTaskManagers;
+ }
- public void setNumTaskManagers(int numTaskManagers) { this.numTaskManagers = numTaskManagers; }
-
// --------------------------------------------------------------------------------------------
// Temporary File Utilities
// --------------------------------------------------------------------------------------------
-
+
public String getTempDirPath(String dirName) throws IOException {
File f = createAndRegisterTempFile(dirName);
return f.toURI().toString();
}
-
+
public String getTempFilePath(String fileName) throws IOException {
File f = createAndRegisterTempFile(fileName);
return f.toURI().toString();
}
-
+
public String createTempFile(String fileName, String contents) throws IOException {
File f = createAndRegisterTempFile(fileName);
Files.write(contents, f, Charsets.UTF_8);
return f.toURI().toString();
}
-
+
public File createAndRegisterTempFile(String fileName) throws IOException {
File baseDir = new File(System.getProperty("java.io.tmpdir"));
File f = new File(baseDir, this.getClass().getName() + "-" + fileName);
-
+
if (f.exists()) {
deleteRecursively(f);
}
-
+
File parentToDelete = f;
while (true) {
File parent = parentToDelete.getParentFile();
@@ -206,12 +131,12 @@ public abstract class AbstractTestBase {
}
parentToDelete = parent;
}
-
+
Files.createParentDirs(f);
this.tempFiles.add(parentToDelete);
return f;
}
-
+
private void deleteAllTempFiles() throws IOException {
for (File f : this.tempFiles) {
if (f.exists()) {
@@ -219,219 +144,4 @@ public abstract class AbstractTestBase {
}
}
}
-
- private static void deleteRecursively (File f) throws IOException {
- if (f.isDirectory()) {
- FileUtils.deleteDirectory(f);
- } else {
- f.delete();
- }
- }
-
- // --------------------------------------------------------------------------------------------
- // Result Checking
- // --------------------------------------------------------------------------------------------
-
- public BufferedReader[] getResultReader(String resultPath) throws IOException {
- return getResultReader(resultPath, new String[]{}, false);
- }
-
- public BufferedReader[] getResultReader(String resultPath, String[] excludePrefixes, boolean inOrderOfFiles) throws IOException {
- File[] files = getAllInvolvedFiles(resultPath, excludePrefixes);
-
- if (inOrderOfFiles) {
- // sort the files after their name (1, 2, 3, 4)...
- // we cannot sort by path, because strings sort by prefix
- Arrays.sort(files, new Comparator<File>() {
-
- @Override
- public int compare(File o1, File o2) {
- try {
- int f1 = Integer.parseInt(o1.getName());
- int f2 = Integer.parseInt(o2.getName());
- return f1 < f2 ? -1 : (f1 > f2 ? 1 : 0);
- }
- catch (NumberFormatException e) {
- throw new RuntimeException("The file names are no numbers and cannot be ordered: " +
- o1.getName() + "/" + o2.getName());
- }
- }
- });
- }
-
- BufferedReader[] readers = new BufferedReader[files.length];
- for (int i = 0; i < files.length; i++) {
- readers[i] = new BufferedReader(new FileReader(files[i]));
- }
- return readers;
- }
-
-
-
- public BufferedInputStream[] getResultInputStream(String resultPath) throws IOException {
- return getResultInputStream(resultPath, new String[]{});
- }
-
- public BufferedInputStream[] getResultInputStream(String resultPath, String[] excludePrefixes) throws IOException {
- File[] files = getAllInvolvedFiles(resultPath, excludePrefixes);
- BufferedInputStream[] inStreams = new BufferedInputStream[files.length];
- for (int i = 0; i < files.length; i++) {
- inStreams[i] = new BufferedInputStream(new FileInputStream(files[i]));
- }
- return inStreams;
- }
-
- public void readAllResultLines(List<String> target, String resultPath) throws IOException {
- readAllResultLines(target, resultPath, new String[]{});
- }
-
- public void readAllResultLines(List<String> target, String resultPath, String[] excludePrefixes) throws IOException {
- readAllResultLines(target, resultPath, excludePrefixes, false);
- }
-
- public void readAllResultLines(List<String> target, String resultPath, String[] excludePrefixes, boolean inOrderOfFiles) throws IOException {
- for (BufferedReader reader : getResultReader(resultPath, excludePrefixes, inOrderOfFiles)) {
- String s = null;
- while ((s = reader.readLine()) != null) {
- target.add(s);
- }
- }
- }
-
- public void compareResultsByLinesInMemory(String expectedResultStr, String resultPath) throws Exception {
- compareResultsByLinesInMemory(expectedResultStr, resultPath, new String[]{});
- }
-
- public void compareResultsByLinesInMemory(String expectedResultStr, String resultPath, String[] excludePrefixes) throws Exception {
- ArrayList<String> list = new ArrayList<String>();
- readAllResultLines(list, resultPath, excludePrefixes, false);
-
- String[] result = (String[]) list.toArray(new String[list.size()]);
- Arrays.sort(result);
-
- String[] expected = expectedResultStr.isEmpty() ? new String[0] : expectedResultStr.split("\n");
- Arrays.sort(expected);
-
- Assert.assertEquals("Different number of lines in expected and obtained result.", expected.length, result.length);
- Assert.assertArrayEquals(expected, result);
- }
- public void compareResultsByLinesInMemoryWithStrictOrder(String expectedResultStr, String resultPath) throws Exception {
- compareResultsByLinesInMemoryWithStrictOrder(expectedResultStr, resultPath, new String[]{});
- }
-
- public void compareResultsByLinesInMemoryWithStrictOrder(String expectedResultStr, String resultPath, String[] excludePrefixes) throws Exception {
- ArrayList<String> list = new ArrayList<String>();
- readAllResultLines(list, resultPath, excludePrefixes, true);
-
- String[] result = (String[]) list.toArray(new String[list.size()]);
-
- String[] expected = expectedResultStr.split("\n");
-
- Assert.assertEquals("Different number of lines in expected and obtained result.", expected.length, result.length);
- Assert.assertArrayEquals(expected, result);
- }
-
- public void compareKeyValueParisWithDelta(String expectedLines, String resultPath, String delimiter, double maxDelta) throws Exception {
- compareKeyValueParisWithDelta(expectedLines, resultPath, new String[]{}, delimiter, maxDelta);
- }
-
- public void compareKeyValueParisWithDelta(String expectedLines, String resultPath, String[] excludePrefixes, String delimiter, double maxDelta) throws Exception {
- ArrayList<String> list = new ArrayList<String>();
- readAllResultLines(list, resultPath, excludePrefixes, false);
-
- String[] result = (String[]) list.toArray(new String[list.size()]);
- String[] expected = expectedLines.isEmpty() ? new String[0] : expectedLines.split("\n");
-
- Assert.assertEquals("Wrong number of result lines.", expected.length, result.length);
-
- Arrays.sort(result);
- Arrays.sort(expected);
-
- for (int i = 0; i < expected.length; i++) {
- String[] expectedFields = expected[i].split(delimiter);
- String[] resultFields = result[i].split(delimiter);
-
- double expectedPayLoad = Double.parseDouble(expectedFields[1]);
- double resultPayLoad = Double.parseDouble(resultFields[1]);
-
- Assert.assertTrue("Values differ by more than the permissible delta", Math.abs(expectedPayLoad - resultPayLoad) < maxDelta);
- }
- }
-
- public static <X> void compareResultCollections(List<X> expected, List<X> actual, Comparator<X> comparator) {
- Assert.assertEquals(expected.size(), actual.size());
-
- Collections.sort(expected, comparator);
- Collections.sort(actual, comparator);
-
- for (int i = 0; i < expected.size(); i++) {
- Assert.assertEquals(expected.get(i), actual.get(i));
- }
- }
-
- private File[] getAllInvolvedFiles(String resultPath, String[] excludePrefixes) {
- final String[] exPrefs = excludePrefixes;
- File result = asFile(resultPath);
- if (!result.exists()) {
- Assert.fail("Result file was not written");
- }
- if (result.isDirectory()) {
- return result.listFiles(new FilenameFilter() {
-
- @Override
- public boolean accept(File dir, String name) {
- for(String p: exPrefs) {
- if(name.startsWith(p)) {
- return false;
- }
- }
- return true;
- }
- });
- } else {
- return new File[] { result };
- }
- }
-
- public File asFile(String path) {
- try {
- URI uri = new URI(path);
- if (uri.getScheme().equals("file")) {
- return new File(uri.getPath());
- } else {
- throw new IllegalArgumentException("This path does not denote a local file.");
- }
- } catch (URISyntaxException e) {
- throw new IllegalArgumentException("This path does not describe a valid local file URI.");
- }
- }
-
- // --------------------------------------------------------------------------------------------
- // Miscellaneous helper methods
- // --------------------------------------------------------------------------------------------
-
- protected static Collection<Object[]> toParameterList(Configuration ... testConfigs) {
- ArrayList<Object[]> configs = new ArrayList<Object[]>();
- for (Configuration testConfig : testConfigs) {
- Object[] c = { testConfig };
- configs.add(c);
- }
- return configs;
- }
-
- protected static Collection<Object[]> toParameterList(List<Configuration> testConfigs) {
- LinkedList<Object[]> configs = new LinkedList<Object[]>();
- for (Configuration testConfig : testConfigs) {
- Object[] c = { testConfig };
- configs.add(c);
- }
- return configs;
- }
-
- public static PrintStream getNullPrintStream() {
- return new PrintStream(new OutputStream() {
- @Override
- public void write(int b) throws IOException {}
- });
- }
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-test-utils/src/main/java/org/apache/flink/test/util/CollectionTestEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/CollectionTestEnvironment.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/CollectionTestEnvironment.java
new file mode 100644
index 0000000..2840914
--- /dev/null
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/CollectionTestEnvironment.java
@@ -0,0 +1,52 @@
+/*
+ * 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.test.util;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.java.CollectionEnvironment;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.ExecutionEnvironmentFactory;
+
+public class CollectionTestEnvironment extends CollectionEnvironment {
+
+ protected JobExecutionResult latestResult;
+
+ @Override
+ public JobExecutionResult execute() throws Exception {
+ return execute("test job");
+ }
+
+ @Override
+ public JobExecutionResult execute(String jobName) throws Exception {
+ JobExecutionResult result = super.execute(jobName);
+ this.latestResult = result;
+ return result;
+ }
+
+ protected void setAsContext() {
+ ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
+ @Override
+ public ExecutionEnvironment createExecutionEnvironment() {
+ return CollectionTestEnvironment.this;
+ }
+ };
+
+ initializeContextEnvironment(factory);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
index 03f60b3..3304c7d 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
@@ -20,22 +20,10 @@ package org.apache.flink.test.util;
import java.util.Comparator;
-import akka.actor.ActorRef;
import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.compiler.DataStatistics;
-import org.apache.flink.compiler.PactCompiler;
-import org.apache.flink.compiler.plan.OptimizedPlan;
-import org.apache.flink.compiler.plandump.PlanJSONDumpGenerator;
-import org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.client.JobClient;
-import org.apache.flink.runtime.jobgraph.JobGraph;
import org.junit.Assert;
import org.junit.Test;
-import org.apache.flink.api.java.CollectionEnvironment;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.ExecutionEnvironmentFactory;
import org.apache.flink.api.java.tuple.Tuple;
@@ -191,97 +179,6 @@ public abstract class JavaProgramTestBase extends AbstractTestBase {
}
}
- private static final class TestEnvironment extends ExecutionEnvironment {
-
- private final ForkableFlinkMiniCluster executor;
-
- private JobExecutionResult latestResult;
-
-
- private TestEnvironment(ForkableFlinkMiniCluster executor, int degreeOfParallelism) {
- this.executor = executor;
- setDegreeOfParallelism(degreeOfParallelism);
- }
-
- @Override
- public JobExecutionResult execute(String jobName) throws Exception {
- try {
- OptimizedPlan op = compileProgram(jobName);
-
- NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
- JobGraph jobGraph = jgg.compileJobGraph(op);
-
- ActorRef client = this.executor.getJobClient();
- JobExecutionResult result = JobClient.submitJobAndWait(jobGraph, false, client);
-
- this.latestResult = result;
- return result;
- }
- catch (Exception e) {
- System.err.println(e.getMessage());
- e.printStackTrace();
- Assert.fail("Job execution failed!");
- return null;
- }
- }
-
-
- @Override
- public String getExecutionPlan() throws Exception {
- OptimizedPlan op = compileProgram("unused");
-
- PlanJSONDumpGenerator jsonGen = new PlanJSONDumpGenerator();
- return jsonGen.getOptimizerPlanAsJSON(op);
- }
-
-
- private OptimizedPlan compileProgram(String jobName) {
- Plan p = createProgramPlan(jobName);
-
- PactCompiler pc = new PactCompiler(new DataStatistics());
- return pc.compile(p);
- }
-
- private void setAsContext() {
- ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
- @Override
- public ExecutionEnvironment createExecutionEnvironment() {
- return TestEnvironment.this;
- }
- };
-
- initializeContextEnvironment(factory);
- }
- }
-
- private static final class CollectionTestEnvironment extends CollectionEnvironment {
-
- private JobExecutionResult latestResult;
-
- @Override
- public JobExecutionResult execute() throws Exception {
- return execute("test job");
- }
-
- @Override
- public JobExecutionResult execute(String jobName) throws Exception {
- JobExecutionResult result = super.execute(jobName);
- this.latestResult = result;
- return result;
- }
-
- private void setAsContext() {
- ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
- @Override
- public ExecutionEnvironment createExecutionEnvironment() {
- return CollectionTestEnvironment.this;
- }
- };
-
- initializeContextEnvironment(factory);
- }
- }
-
public static class TupleComparator<T extends Tuple> implements Comparator<T> {
@SuppressWarnings("unchecked")
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-test-utils/src/main/java/org/apache/flink/test/util/MultipleProgramsTestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/MultipleProgramsTestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/MultipleProgramsTestBase.java
new file mode 100644
index 0000000..225936c
--- /dev/null
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/MultipleProgramsTestBase.java
@@ -0,0 +1,66 @@
+/*
+ * 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.test.util;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+public class MultipleProgramsTestBase extends TestBaseUtils {
+ protected static ForkableFlinkMiniCluster cluster = null;
+ protected transient ExecutionMode mode;
+
+ public MultipleProgramsTestBase(ExecutionMode mode){
+ this.mode = mode;
+ switch(mode){
+ case CLUSTER:
+ TestEnvironment clusterEnv = new TestEnvironment(cluster, 4);
+ clusterEnv.setAsContext();
+ break;
+ case COLLECTION:
+ CollectionTestEnvironment collectionEnv = new CollectionTestEnvironment();
+ collectionEnv.setAsContext();
+ break;
+ }
+ }
+
+ @BeforeClass
+ public static void setup() throws Exception{
+ cluster = TestBaseUtils.startCluster(1, 4);
+
+ }
+
+ @AfterClass
+ public static void teardown() throws Exception {
+ stopCluster(cluster, TestBaseUtils.DEFAULT_TIMEOUT);
+ }
+
+ @Parameterized.Parameters(name = "Execution mode = {0}")
+ public static Collection<ExecutionMode[]> executionModes(){
+ return Arrays.asList(new ExecutionMode[]{ExecutionMode.CLUSTER},
+ new ExecutionMode[]{ExecutionMode.COLLECTION});
+ }
+
+ protected static enum ExecutionMode{
+ CLUSTER, COLLECTION
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/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
new file mode 100644
index 0000000..7939a3c
--- /dev/null
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
@@ -0,0 +1,353 @@
+/*
+ * 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.test.util;
+
+import akka.actor.ActorRef;
+import akka.dispatch.Futures;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import org.apache.commons.io.FileUtils;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
+import org.apache.hadoop.fs.FileSystem;
+import org.junit.Assert;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileReader;
+import java.io.FilenameFilter;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class TestBaseUtils {
+ protected static final int MINIMUM_HEAP_SIZE_MB = 192;
+
+ protected static final long TASK_MANAGER_MEMORY_SIZE = 80;
+
+ protected static final int DEFAULT_TASK_MANAGER_NUM_SLOTS = 1;
+
+ protected static final int DEFAULT_NUM_TASK_MANAGERS = 1;
+
+ protected static final int DEFAULT_AKKA_ASK_TIMEOUT = 1000;
+
+ protected static final String DEFAULT_AKKA_STARTUP_TIMEOUT = "60 s";
+
+ protected static FiniteDuration DEFAULT_TIMEOUT = new FiniteDuration
+ (DEFAULT_AKKA_ASK_TIMEOUT, TimeUnit.SECONDS);
+
+ protected TestBaseUtils(){
+ verifyJvmOptions();
+ }
+
+ private void verifyJvmOptions() {
+ long heap = Runtime.getRuntime().maxMemory() >> 20;
+ Assert.assertTrue("Insufficient java heap space " + heap + "mb - set JVM option: -Xmx" + MINIMUM_HEAP_SIZE_MB
+ + "m", heap > MINIMUM_HEAP_SIZE_MB - 50);
+ }
+
+ protected static ForkableFlinkMiniCluster startCluster(int numTaskManagers, int
+ taskManagerNumSlots) {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, true);
+ config.setBoolean(ConfigConstants.TASK_MANAGER_MEMORY_LAZY_ALLOCATION_KEY, true);
+ config.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, TASK_MANAGER_MEMORY_SIZE);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, taskManagerNumSlots);
+ config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers);
+ config.setInteger(ConfigConstants.AKKA_ASK_TIMEOUT, DEFAULT_AKKA_ASK_TIMEOUT);
+ config.setString(ConfigConstants.AKKA_STARTUP_TIMEOUT, DEFAULT_AKKA_STARTUP_TIMEOUT);
+ return new ForkableFlinkMiniCluster(config);
+ }
+
+ protected static void stopCluster(ForkableFlinkMiniCluster executor, FiniteDuration timeout)
+ throws Exception {
+ if(executor != null) {
+ int numUnreleasedBCVars = 0;
+ {
+ List<ActorRef> tms = executor.getTaskManagersAsJava();
+ List<Future<Object>> responseFutures = new ArrayList<Future<Object>>();
+
+ for (ActorRef tm : tms) {
+ responseFutures.add(Patterns.ask(tm, TestingTaskManagerMessages
+ .RequestBroadcastVariablesWithReferences$.MODULE$, new Timeout
+ (timeout)));
+ }
+
+ Future<Iterable<Object>> futureResponses = Futures.sequence(
+ responseFutures, AkkaUtils.globalExecutionContext());
+
+ Iterable<Object> responses = Await.result(futureResponses, timeout);
+
+ for (Object response : responses) {
+ numUnreleasedBCVars += ((TestingTaskManagerMessages
+ .ResponseBroadcastVariablesWithReferences) response).number();
+ }
+ }
+
+ executor.stop();
+ FileSystem.closeAll();
+ System.gc();
+
+ Assert.assertEquals("Not all broadcast variables were released.", 0, numUnreleasedBCVars);
+ }
+
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Result Checking
+ // --------------------------------------------------------------------------------------------
+
+ public BufferedReader[] getResultReader(String resultPath) throws IOException {
+ return getResultReader(resultPath, new String[]{}, false);
+ }
+
+ public BufferedReader[] getResultReader(String resultPath, String[] excludePrefixes,
+ boolean inOrderOfFiles) throws IOException {
+ File[] files = getAllInvolvedFiles(resultPath, excludePrefixes);
+
+ if (inOrderOfFiles) {
+ // sort the files after their name (1, 2, 3, 4)...
+ // we cannot sort by path, because strings sort by prefix
+ Arrays.sort(files, new Comparator<File>() {
+
+ @Override
+ public int compare(File o1, File o2) {
+ try {
+ int f1 = Integer.parseInt(o1.getName());
+ int f2 = Integer.parseInt(o2.getName());
+ return f1 < f2 ? -1 : (f1 > f2 ? 1 : 0);
+ }
+ catch (NumberFormatException e) {
+ throw new RuntimeException("The file names are no numbers and cannot be ordered: " +
+ o1.getName() + "/" + o2.getName());
+ }
+ }
+ });
+ }
+
+ BufferedReader[] readers = new BufferedReader[files.length];
+ for (int i = 0; i < files.length; i++) {
+ readers[i] = new BufferedReader(new FileReader(files[i]));
+ }
+ return readers;
+ }
+
+
+
+ public BufferedInputStream[] getResultInputStream(String resultPath) throws
+ IOException {
+ return getResultInputStream(resultPath, new String[]{});
+ }
+
+ public BufferedInputStream[] getResultInputStream(String resultPath, String[]
+ excludePrefixes) throws IOException {
+ File[] files = getAllInvolvedFiles(resultPath, excludePrefixes);
+ BufferedInputStream[] inStreams = new BufferedInputStream[files.length];
+ for (int i = 0; i < files.length; i++) {
+ inStreams[i] = new BufferedInputStream(new FileInputStream(files[i]));
+ }
+ return inStreams;
+ }
+
+ public void readAllResultLines(List<String> target, String resultPath) throws
+ IOException {
+ readAllResultLines(target, resultPath, new String[]{});
+ }
+
+ public void readAllResultLines(List<String> target, String resultPath, String[]
+ excludePrefixes) throws IOException {
+ readAllResultLines(target, resultPath, excludePrefixes, false);
+ }
+
+ public void readAllResultLines(List<String> target, String resultPath, String[]
+ excludePrefixes, boolean inOrderOfFiles) throws IOException {
+ for (BufferedReader reader : getResultReader(resultPath, excludePrefixes, inOrderOfFiles)) {
+ String s = null;
+ while ((s = reader.readLine()) != null) {
+ target.add(s);
+ }
+ }
+ }
+
+ public void compareResultsByLinesInMemory(String expectedResultStr, String
+ resultPath) throws Exception {
+ compareResultsByLinesInMemory(expectedResultStr, resultPath, new String[]{});
+ }
+
+ public void compareResultsByLinesInMemory(String expectedResultStr, String resultPath,
+ String[] excludePrefixes) throws Exception {
+ ArrayList<String> list = new ArrayList<String>();
+ readAllResultLines(list, resultPath, excludePrefixes, false);
+
+ String[] result = (String[]) list.toArray(new String[list.size()]);
+ Arrays.sort(result);
+
+ String[] expected = expectedResultStr.isEmpty() ? new String[0] : expectedResultStr.split("\n");
+ Arrays.sort(expected);
+
+ Assert.assertEquals("Different number of lines in expected and obtained result.", expected.length, result.length);
+ Assert.assertArrayEquals(expected, result);
+ }
+
+ public void compareResultsByLinesInMemoryWithStrictOrder(String expectedResultStr,
+ String resultPath) throws
+ Exception {
+ compareResultsByLinesInMemoryWithStrictOrder(expectedResultStr, resultPath, new String[]{});
+ }
+
+ public void compareResultsByLinesInMemoryWithStrictOrder(String expectedResultStr,
+ String resultPath, String[] excludePrefixes) throws Exception {
+ ArrayList<String> list = new ArrayList<String>();
+ readAllResultLines(list, resultPath, excludePrefixes, true);
+
+ String[] result = (String[]) list.toArray(new String[list.size()]);
+
+ String[] expected = expectedResultStr.split("\n");
+
+ Assert.assertEquals("Different number of lines in expected and obtained result.", expected.length, result.length);
+ Assert.assertArrayEquals(expected, result);
+ }
+
+ public void compareKeyValueParisWithDelta(String expectedLines, String resultPath,
+ String delimiter, double maxDelta) throws Exception {
+ compareKeyValueParisWithDelta(expectedLines, resultPath, new String[]{}, delimiter, maxDelta);
+ }
+
+ public void compareKeyValueParisWithDelta(String expectedLines, String resultPath,
+ String[] excludePrefixes, String delimiter, double maxDelta) throws Exception {
+ ArrayList<String> list = new ArrayList<String>();
+ readAllResultLines(list, resultPath, excludePrefixes, false);
+
+ String[] result = (String[]) list.toArray(new String[list.size()]);
+ String[] expected = expectedLines.isEmpty() ? new String[0] : expectedLines.split("\n");
+
+ Assert.assertEquals("Wrong number of result lines.", expected.length, result.length);
+
+ Arrays.sort(result);
+ Arrays.sort(expected);
+
+ for (int i = 0; i < expected.length; i++) {
+ String[] expectedFields = expected[i].split(delimiter);
+ String[] resultFields = result[i].split(delimiter);
+
+ double expectedPayLoad = Double.parseDouble(expectedFields[1]);
+ double resultPayLoad = Double.parseDouble(resultFields[1]);
+
+ Assert.assertTrue("Values differ by more than the permissible delta", Math.abs(expectedPayLoad - resultPayLoad) < maxDelta);
+ }
+ }
+
+ public <X> void compareResultCollections(List<X> expected, List<X> actual,
+ Comparator<X> comparator) {
+ Assert.assertEquals(expected.size(), actual.size());
+
+ Collections.sort(expected, comparator);
+ Collections.sort(actual, comparator);
+
+ for (int i = 0; i < expected.size(); i++) {
+ Assert.assertEquals(expected.get(i), actual.get(i));
+ }
+ }
+
+ private static File[] getAllInvolvedFiles(String resultPath, String[] excludePrefixes) {
+ final String[] exPrefs = excludePrefixes;
+ File result = asFile(resultPath);
+ if (!result.exists()) {
+ Assert.fail("Result file was not written");
+ }
+ if (result.isDirectory()) {
+ return result.listFiles(new FilenameFilter() {
+
+ @Override
+ public boolean accept(File dir, String name) {
+ for(String p: exPrefs) {
+ if(name.startsWith(p)) {
+ return false;
+ }
+ }
+ return true;
+ }
+ });
+ } else {
+ return new File[] { result };
+ }
+ }
+
+ protected static File asFile(String path) {
+ try {
+ URI uri = new URI(path);
+ if (uri.getScheme().equals("file")) {
+ return new File(uri.getPath());
+ } else {
+ throw new IllegalArgumentException("This path does not denote a local file.");
+ }
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException("This path does not describe a valid local file URI.");
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Miscellaneous helper methods
+ // --------------------------------------------------------------------------------------------
+
+ protected static Collection<Object[]> toParameterList(Configuration ... testConfigs) {
+ ArrayList<Object[]> configs = new ArrayList<Object[]>();
+ for (Configuration testConfig : testConfigs) {
+ Object[] c = { testConfig };
+ configs.add(c);
+ }
+ return configs;
+ }
+
+ protected static Collection<Object[]> toParameterList(List<Configuration> testConfigs) {
+ LinkedList<Object[]> configs = new LinkedList<Object[]>();
+ for (Configuration testConfig : testConfigs) {
+ Object[] c = { testConfig };
+ configs.add(c);
+ }
+ return configs;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // File helper methods
+ // --------------------------------------------------------------------------------------------
+
+ protected static void deleteRecursively(File f) throws IOException {
+ if (f.isDirectory()) {
+ FileUtils.deleteDirectory(f);
+ } else {
+ f.delete();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
new file mode 100644
index 0000000..439f8f6
--- /dev/null
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
@@ -0,0 +1,96 @@
+/*
+ * 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.test.util;
+
+import akka.actor.ActorRef;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.ExecutionEnvironmentFactory;
+import org.apache.flink.compiler.DataStatistics;
+import org.apache.flink.compiler.PactCompiler;
+import org.apache.flink.compiler.plan.OptimizedPlan;
+import org.apache.flink.compiler.plandump.PlanJSONDumpGenerator;
+import org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator;
+import org.apache.flink.runtime.client.JobClient;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.junit.Assert;
+
+public class TestEnvironment extends ExecutionEnvironment {
+
+ private final ForkableFlinkMiniCluster executor;
+
+ protected JobExecutionResult latestResult;
+
+
+ public TestEnvironment(ForkableFlinkMiniCluster executor, int degreeOfParallelism) {
+ this.executor = executor;
+ setDegreeOfParallelism(degreeOfParallelism);
+ }
+
+ @Override
+ public JobExecutionResult execute(String jobName) throws Exception {
+ try {
+ OptimizedPlan op = compileProgram(jobName);
+
+ NepheleJobGraphGenerator jgg = new NepheleJobGraphGenerator();
+ JobGraph jobGraph = jgg.compileJobGraph(op);
+
+ ActorRef client = this.executor.getJobClient();
+ JobExecutionResult result = JobClient.submitJobAndWait(jobGraph, false, client);
+
+ this.latestResult = result;
+ return result;
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail("Job execution failed!");
+ return null;
+ }
+ }
+
+
+ @Override
+ public String getExecutionPlan() throws Exception {
+ OptimizedPlan op = compileProgram("unused");
+
+ PlanJSONDumpGenerator jsonGen = new PlanJSONDumpGenerator();
+ return jsonGen.getOptimizerPlanAsJSON(op);
+ }
+
+
+ private OptimizedPlan compileProgram(String jobName) {
+ Plan p = createProgramPlan(jobName);
+
+ PactCompiler pc = new PactCompiler(new DataStatistics());
+ return pc.compile(p);
+ }
+
+ protected void setAsContext() {
+ ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
+ @Override
+ public ExecutionEnvironment createExecutionEnvironment() {
+ return TestEnvironment.this;
+ }
+ };
+
+ initializeContextEnvironment(factory);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index 8761ec6..1ebe9eb 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -46,7 +46,6 @@ LocalFlinkMiniCluster(userConfiguration) {
config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, jobManagerRPC)
config.setInteger(ConfigConstants.TASK_MANAGER_IPC_PORT_KEY, taskManagerRPC)
config.setInteger(ConfigConstants.TASK_MANAGER_DATA_PORT_KEY, taskManagerData)
-
}
super.generateConfiguration(config)
@@ -68,8 +67,14 @@ LocalFlinkMiniCluster(userConfiguration) {
config.setInteger(ConfigConstants.TASK_MANAGER_DATA_PORT_KEY, dataPort + index)
}
+ val localExecution = if(numTaskManagers == 1){
+ false
+ }else{
+ false
+ }
+
val (connectionInfo, jobManagerAkkaURL, taskManagerConfig, networkConnectionConfig) =
- TaskManager.parseConfiguration(HOSTNAME, config, false)
+ TaskManager.parseConfiguration(HOSTNAME, config, localExecution)
system.actorOf(Props(new TaskManager(connectionInfo, jobManagerAkkaURL, taskManagerConfig,
networkConnectionConfig) with TestingTaskManager), TaskManager.TASK_MANAGER_NAME + index)
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/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 3409c17..0c3ad4a 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
@@ -18,79 +18,63 @@
package org.apache.flink.test.exampleJavaPrograms;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.LinkedList;
-import org.apache.flink.configuration.Configuration;
+import java.io.File;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
import org.apache.flink.examples.java.graph.PageRankBasic;
import org.apache.flink.test.testdata.PageRankData;
-import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
-public class PageRankITCase extends JavaProgramTestBase {
+public class PageRankITCase extends MultipleProgramsTestBase {
+
+ public PageRankITCase(ExecutionMode mode){
+ super(mode);
+ }
- private int curProgId = config.getInteger("ProgramId", -1);
-
private String verticesPath;
private String edgesPath;
private String resultPath;
- private String expectedResult;
-
- public PageRankITCase(Configuration config) {
- super(config);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
- verticesPath = createTempFile("vertices.txt", PageRankData.VERTICES);
- edgesPath = createTempFile("edges.txt", PageRankData.EDGES);
- }
+ private String expected;
- @Override
- protected void testProgram() throws Exception {
- expectedResult = runProgram(curProgId);
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareKeyValueParisWithDelta(expectedResult, resultPath, " ", 0.01);
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws Exception{
+ resultPath = tempFolder.newFile().toURI().toString();
+ File verticesFile = tempFolder.newFile();
+ Files.write(PageRankData.VERTICES, verticesFile, Charsets.UTF_8);
+
+ File edgesFile = tempFolder.newFile();
+ Files.write(PageRankData.EDGES, edgesFile, Charsets.UTF_8);
+
+ verticesPath = verticesFile.toURI().toString();
+ edgesPath = edgesFile.toURI().toString();
}
-
- @Parameters
- public static Collection<Object[]> getConfigurations() throws IOException {
- LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+ @After
+ public void after() throws Exception{
+ compareKeyValueParisWithDelta(expected, resultPath, " ", 0.01);
+ }
- int NUM_PROGRAMS = 2;
- for(int i=1; i <= NUM_PROGRAMS; i++) {
- Configuration config = new Configuration();
- config.setInteger("ProgramId", i);
- tConfigs.add(config);
- }
-
- return toParameterList(tConfigs);
+ @Test
+ public void testPageRankSmallNumberOfIterations() throws Exception{
+ PageRankBasic.main(new String[] {verticesPath, edgesPath, resultPath, PageRankData.NUM_VERTICES+"", "3"});
+ expected = PageRankData.RANKS_AFTER_3_ITERATIONS;
}
-
- public String runProgram(int progId) throws Exception {
-
- switch(progId) {
- case 1: {
- PageRankBasic.main(new String[] {verticesPath, edgesPath, resultPath, PageRankData.NUM_VERTICES+"", "3"});
- return PageRankData.RANKS_AFTER_3_ITERATIONS;
- }
- case 2: {
- // start with a very high number of iteration such that the dynamic convergence criterion must handle termination
- PageRankBasic.main(new String[] {verticesPath, edgesPath, resultPath, PageRankData.NUM_VERTICES+"", "1000"});
- return PageRankData.RANKS_AFTER_EPSILON_0_0001_CONVERGENCE;
- }
-
- default:
- throw new IllegalArgumentException("Invalid program id");
- }
+ @Test
+ public void testPageRankWithConvergenceCriterion() throws Exception {
+ PageRankBasic.main(new String[] {verticesPath, edgesPath, resultPath, PageRankData.NUM_VERTICES+"", "1000"});
+ expected = PageRankData.RANKS_AFTER_EPSILON_0_0001_CONVERGENCE;
}
}