[FLINK-1878] [streaming] Stream environments accept a flag that controls...

[FLINK-1878] [streaming] Stream environments accept a flag that controls sysout logging during execution.

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java

@@ -22,8 +22,6 @@ import org.apache.flink.streaming.util.ClusterUtil;
 
 public class LocalStreamEnvironment extends StreamExecutionEnvironment {
 
-	protected static ClassLoader userClassLoader;
-
 	/**
 	 * Executes the JobGraph of the on a mini cluster of CLusterUtil with a
 	 * default name.
@@ -32,7 +30,8 @@ public class LocalStreamEnvironment extends StreamExecutionEnvironment {
 	 */
 	@Override
 	public JobExecutionResult execute() throws Exception {
-		return ClusterUtil.runOnMiniCluster(this.streamGraph.getJobGraph(), getParallelism());
+		return ClusterUtil.runOnMiniCluster(this.streamGraph.getJobGraph(), getParallelism(),
+																getConfig().isSysoutLoggingEnabled());
 	}
 
 	/**
@@ -45,6 +44,7 @@ public class LocalStreamEnvironment extends StreamExecutionEnvironment {
 	 */
 	@Override
 	public JobExecutionResult execute(String jobName) throws Exception {
-		return ClusterUtil.runOnMiniCluster(this.streamGraph.getJobGraph(jobName), getParallelism());
+		return ClusterUtil.runOnMiniCluster(this.streamGraph.getJobGraph(jobName), getParallelism(),
+																		getConfig().isSysoutLoggingEnabled());
 	}
 }

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java

@@ -69,11 +69,12 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
 		this.host = host;
 		this.port = port;
 		this.jarFiles = new ArrayList<File>();
-		for (int i = 0; i < jarFiles.length; i++) {
-			File file = new File(jarFiles[i]);
+		for (String jarFile : jarFiles) {
+			File file = new File(jarFile);
 			try {
 				JobWithJars.checkJarFile(file);
-			} catch (IOException e) {
+			}
+			catch (IOException e) {
 				throw new RuntimeException("Problem with jar file " + file.getAbsolutePath(), e);
 			}
 			this.jarFiles.add(file);
@@ -113,6 +114,7 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
 		Configuration configuration = jobGraph.getJobConfiguration();
 		Client client = new Client(new InetSocketAddress(host, port), configuration,
 				JobWithJars.buildUserCodeClassLoader(jarFiles, JobWithJars.class.getClassLoader()), -1);
+		client.setPrintStatusDuringExecution(getConfig().isSysoutLoggingEnabled());
 		
 		try {
 			JobSubmissionResult result = client.run(jobGraph, true);

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/util/ClusterUtil.java

@@ -42,9 +42,8 @@ public class ClusterUtil {
 	 *            memorySize
 	 * @return The result of the job execution, containing elapsed time and accumulators.
 	 */
-	public static JobExecutionResult runOnMiniCluster(JobGraph jobGraph, int parallelism, long memorySize)
-			throws Exception {
-
+	public static JobExecutionResult runOnMiniCluster(JobGraph jobGraph, int parallelism, long memorySize,
+																boolean printDuringExecution) throws Exception {
 		Configuration configuration = jobGraph.getJobConfiguration();
 
 		LocalFlinkMiniCluster exec = null;
@@ -58,7 +57,7 @@ public class ClusterUtil {
 		try {
 			exec = new LocalFlinkMiniCluster(configuration, true);
 
-			SerializedJobExecutionResult result = exec.submitJobAndWait(jobGraph, true);
+			SerializedJobExecutionResult result = exec.submitJobAndWait(jobGraph, printDuringExecution);
 			return result.toJobExecutionResult(ClusterUtil.class.getClassLoader());
 		}
 		finally {
@@ -68,7 +67,8 @@ public class ClusterUtil {
 		}
 	}
 
-	public static JobExecutionResult runOnMiniCluster(JobGraph jobGraph, int numOfSlots) throws Exception {
-		return runOnMiniCluster(jobGraph, numOfSlots, -1);
+	public static JobExecutionResult runOnMiniCluster(JobGraph jobGraph, int numOfSlots,
+														boolean printDuringExecution) throws Exception {
+		return runOnMiniCluster(jobGraph, numOfSlots, -1, printDuringExecution);
 	}
 }

flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointingITCase.java

@@ -100,6 +100,7 @@ public class StreamCheckpointingITCase {
 																	"localhost", cluster.getJobManagerRPCPort());
 			env.setParallelism(PARALLELISM);
 			env.enableCheckpointing(200);
+			env.getConfig().disableSysoutLogging();
 
 			DataStream<String> stream = env.addSource(new RichParallelSourceFunction<String>() {