[streaming] streamcollector2 bugfix

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector2.java

@@ -47,8 +47,10 @@ public class StreamCollector2<T extends Tuple> implements Collector<T> {
 		this.keyPostition = keyPosition;
 
 		for (int i = 0; i < batchSizesOfNotPartitioned.size(); i++) {
+			List<RecordWriter<StreamRecord>> output = new ArrayList<RecordWriter<StreamRecord>>();
+			output.add(notPartitionedOutputs.get(i));
 			notPartitionedCollectors.add(new StreamCollector<Tuple>(batchSizesOfNotPartitioned.get(i),
-					batchTimeout, channelID, serializationDelegate, notPartitionedOutputs));
+					batchTimeout, channelID, serializationDelegate, output));
 		}
 
 		for (int i = 0; i < batchSizesOfPartitioned.size(); i++) {

flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/PrintTest.java

@@ -51,6 +51,7 @@ public class PrintTest {
 		StreamExecutionEnvironment env = new StreamExecutionEnvironment();
 		DataStream<Tuple2<Integer, String>> source = env.addSource(new MySource(), 1);
 		DataStream<Tuple2<Integer, String>> map = source.flatMap(new MyFlatMap(), 1).print();
+		source.print();
 		env.execute();
 
 	}