[FLINK-2448] [tests] Create new Test Environments on every call to...

[FLINK-2448] [tests] Create new Test Environments on every call to getExecutionEnvironment and allow sharing of job execution results

flink-test-utils/src/main/java/org/apache/flink/test/util/CollectionTestEnvironment.java

@@ -25,6 +25,18 @@ import org.apache.flink.api.java.ExecutionEnvironmentFactory;
 
 public class CollectionTestEnvironment extends CollectionEnvironment {
 
+	private CollectionTestEnvironment lastEnv = null;
+
+	@Override
+	public JobExecutionResult getLastJobExecutionResult() {
+		if (lastEnv == null) {
+			return this.lastJobExecutionResult;
+		}
+		else {
+			return lastEnv.getLastJobExecutionResult();
+		}
+	}
+
 	@Override
 	public JobExecutionResult execute() throws Exception {
 		return execute("test job");
@@ -41,7 +53,8 @@ public class CollectionTestEnvironment extends CollectionEnvironment {
 		ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
 			@Override
 			public ExecutionEnvironment createExecutionEnvironment() {
-				return CollectionTestEnvironment.this;
+				lastEnv = new CollectionTestEnvironment();
+				return lastEnv;
 			}
 		};
 

flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java

@@ -35,6 +35,18 @@ public class TestEnvironment extends ExecutionEnvironment {
 
 	private final ForkableFlinkMiniCluster executor;
 
+	private TestEnvironment lastEnv = null;
+
+	@Override
+	public JobExecutionResult getLastJobExecutionResult() {
+		if (lastEnv == null) {
+			return this.lastJobExecutionResult;
+		}
+		else {
+			return lastEnv.getLastJobExecutionResult();
+		}
+	}
+
 	public TestEnvironment(ForkableFlinkMiniCluster executor, int parallelism) {
 		this.executor = executor;
 		setParallelism(parallelism);
@@ -82,7 +94,8 @@ public class TestEnvironment extends ExecutionEnvironment {
 		ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
 			@Override
 			public ExecutionEnvironment createExecutionEnvironment() {
-				return TestEnvironment.this;
+				lastEnv = new TestEnvironment(executor, getParallelism());
+				return lastEnv;
 			}
 		};