diff --git a/flink-addons/flink-streaming/pom.xml b/flink-addons/flink-streaming/pom.xml
index 25d69b057f6bb6e1edbedfb91c483e1ce77b1bf2..3dd0f179b0f5fc6fc6128138e96602857ebe0f7a 100644
--- a/flink-addons/flink-streaming/pom.xml
+++ b/flink-addons/flink-streaming/pom.xml
@@ -94,12 +94,6 @@
jblas
1.2.3
-
- org.mockito
- mockito-all
- 1.8.5
- test
-
diff --git a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector.java b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector.java
index 5c8b58869753ec221cb574a48aae8908c9d882d5..209e48d8f7664dcdabb024092ef8097a6ccccccc 100755
--- a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector.java
+++ b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector.java
@@ -54,7 +54,7 @@ public class StreamCollector implements Collector {
@Override
public void collect(T tuple) {
//TODO: move copy to StreamCollector2
- streamRecord.setTuple(counter, tuple);
+ streamRecord.setTuple(counter, StreamRecord.copyTuple(tuple));
counter++;
if (counter >= batchSize) {
diff --git a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector2.java b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector2.java
index 5d3468097b11d7b9365214c0c767bdcb8e60ae0e..ae70b5a7b41d051d93719c99effe002bf11d4b78 100644
--- a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector2.java
+++ b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamCollector2.java
@@ -65,17 +65,15 @@ public class StreamCollector2 implements Collector {
// TODO copy here instead of copying inside every StreamCollector
@Override
- public void collect(T tuple) {
- T copiedTuple = StreamRecord.copyTuple(tuple);
-
+ public void collect(T record) {
for (StreamCollector collector : notPartitionedCollectors) {
- collector.collect(copiedTuple);
+ collector.collect(record);
}
- int partitionHash = Math.abs(copiedTuple.getField(keyPostition).hashCode());
+ int partitionHash = Math.abs(record.getField(keyPostition).hashCode());
for (StreamCollector[] collectors : partitionedCollectors) {
- collectors[partitionHash % collectors.length].collect(copiedTuple);
+ collectors[partitionHash % collectors.length].collect(record);
}
}
diff --git a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamWindowTask.java b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamWindowTask.java
index 2d47a20ac16e182450612ccd88c1eb9722e8ae98..dc7b3a21536aff46e2b42d2b445ee352a85ae372 100644
--- a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamWindowTask.java
+++ b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamWindowTask.java
@@ -53,7 +53,7 @@ public class StreamWindowTask extends FlatMapFunction {
@Override
public void flatMap(Tuple value, Collector out) throws Exception {
- long progress = (Long) value.getField(windowFieldId);
+ long progress = value.getField(windowFieldId);
if (initTimestamp == -1) {
initTimestamp = progress;
nextTimestamp = initTimestamp + computeGranularity;
diff --git a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamrecord/StreamRecord.java b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamrecord/StreamRecord.java
index a623d36c1b2236e4bbe5a1d2a5f92cdd0ecfe6b3..e9e8ec521b169915d61fd510c59b8509223876d0 100644
--- a/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamrecord/StreamRecord.java
+++ b/flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamrecord/StreamRecord.java
@@ -147,12 +147,12 @@ public abstract class StreamRecord implements IOReadableWritable, Serializable {
* Tuple to copy
* @return Copy of the tuple
*/
- public static T copyTuple(T tuple) {
+ public static Tuple copyTuple(Tuple tuple) {
// TODO: implement deep copy for arrays
int numofFields = tuple.getArity();
- T newTuple = null;
+ Tuple newTuple = null;
try {
- newTuple = (T) CLASSES[numofFields - 1].newInstance();
+ newTuple = (Tuple) CLASSES[numofFields - 1].newInstance();
for (int i = 0; i < numofFields; i++) {
Class extends Object> type = tuple.getField(i).getClass();
diff --git a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/FlatMapTest.java b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/FlatMapTest.java
index dc0f183778a07b6d900b06ed87a82ee5e822f2ec..36c756f85036eda54aa47c78ad267a860f2cafa9 100644
--- a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/FlatMapTest.java
+++ b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/FlatMapTest.java
@@ -20,6 +20,8 @@ import static org.junit.Assert.fail;
import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
+import java.util.ArrayList;
+import java.util.List;
import org.junit.Test;
@@ -33,6 +35,8 @@ import eu.stratosphere.nephele.jobgraph.AbstractJobVertex;
import eu.stratosphere.nephele.jobgraph.JobInputVertex;
import eu.stratosphere.nephele.jobgraph.JobOutputVertex;
import eu.stratosphere.nephele.jobgraph.JobTaskVertex;
+import eu.stratosphere.streaming.api.MapTest.MyMap;
+import eu.stratosphere.streaming.api.MapTest.MySink;
import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
@@ -40,129 +44,61 @@ import eu.stratosphere.util.Collector;
public class FlatMapTest {
- public static final class MyFlatMap extends FlatMapFunction, Tuple1> {
+ public static final class MyFlatMap extends FlatMapFunction, Tuple1> {
+
@Override
- public void flatMap(Tuple1 value, Collector> out) throws Exception {
- out.collect(value);
- System.out.println("flatMap");
+ public void flatMap(Tuple1 value,
+ Collector> out) throws Exception {
+ out.collect(new Tuple1(value.f0*value.f0));
+
}
+
}
- public static final class MySink extends SinkFunction> {
- int c=0;
+ public static final class MySink extends SinkFunction> {
+
@Override
- public void invoke(Tuple1 tuple) {
- System.out.println(tuple);
- c++;
- System.out.println(c);
+ public void invoke(Tuple1 tuple) {
+ result.add(tuple.f0);
+ System.out.println("result " + tuple.f0);
}
}
- public static final class MySource extends SourceFunction> {
+ public static final class MySource extends SourceFunction> {
@Override
- public void invoke(Collector> collector) {
- for (int i = 0; i < 5; i++) {
- collector.collect(new Tuple1("hi"));
+ public void invoke(Collector> collector)
+ throws Exception {
+ for(int i=0; i<10; i++){
+ collector.collect(new Tuple1(i));
}
}
}
+
+ private static void fillExpectedList(){
+ for(int i=0;i<10;i++){
+ expected.add(i*i);
+ System.out.println("expected " + i*i);
+ }
+ }
- private static final int PARALELISM = 2;
+ private static final int PARALELISM = 1;
+ private static List expected = new ArrayList();
+ private static List result = new ArrayList();
@Test
public void test() throws Exception {
-
- try {
- StreamExecutionEnvironment context2 = new StreamExecutionEnvironment(0, 1000);
- fail();
- } catch (IllegalArgumentException e) {
- try {
- StreamExecutionEnvironment context2 = new StreamExecutionEnvironment(1, 0);
- fail();
- } catch (IllegalArgumentException e2) {
- }
- }
StreamExecutionEnvironment context = new StreamExecutionEnvironment(2, 1000);
- DataStream> dataStream0 = context.addSource(new MySource(),1);
+ DataStream> dataStream0 = context.addSource(new MySource(),1).flatMap(new MyFlatMap(), PARALELISM).addSink(new MySink());
- DataStream> dataStream1 = context.addDummySource().connectWith(dataStream0)
- .partitionBy(0).flatMap(new MyFlatMap(), PARALELISM).broadcast().addSink(new MySink());
context.execute();
-
- JobGraphBuilder jgb = context.jobGB();
-
- for (AbstractJobVertex c : jgb.components.values()) {
- if (c instanceof JobTaskVertex) {
- Configuration config = c.getConfiguration();
- System.out.println(config.getString("componentName", "default"));
- byte[] bytes = config.getBytes("operator", null);
-
- ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
-
- FlatMapFunction f = (FlatMapFunction) in.readObject();
-
- StreamCollector s = new StreamCollector(1, 1000, 1, null);
- Tuple t = new Tuple1("asd");
-
- f.flatMap(t, s);
-
- System.out.println(f.getClass().getGenericSuperclass());
- TupleTypeInfo ts = (TupleTypeInfo) TypeExtractor.createTypeInfo(
- FlatMapFunction.class, f.getClass(), 0, null, null);
-
- System.out.println(ts);
-
- byte[] userFunctionSerialized = config.getBytes("serializedudf", null);
- in = new ObjectInputStream(new ByteArrayInputStream(userFunctionSerialized));
- UserTaskInvokable userFunction = (UserTaskInvokable) in.readObject();
- System.out.println(userFunction.getClass());
- assertTrue(true);
- System.out.println("----------------");
- }
-
- if (c instanceof JobOutputVertex) {
- Configuration config = c.getConfiguration();
- System.out.println(config.getString("componentName", "default"));
- byte[] bytes = config.getBytes("operator", null);
-
- ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
-
- SinkFunction f = (SinkFunction) in.readObject();
-
- System.out.println(f.getClass().getGenericSuperclass());
- TupleTypeInfo ts = (TupleTypeInfo) TypeExtractor.createTypeInfo(
- SinkFunction.class, f.getClass(), 0, null, null);
-
- System.out.println(ts);
-
- byte[] userFunctionSerialized = config.getBytes("serializedudf", null);
- in = new ObjectInputStream(new ByteArrayInputStream(userFunctionSerialized));
- UserSinkInvokable userFunction = (UserSinkInvokable) in.readObject();
- System.out.println(userFunction.getClass());
- assertTrue(true);
- System.out.println("----------------");
- }
-
- if (c instanceof JobInputVertex) {
- Configuration config = c.getConfiguration();
- System.out.println(config.getString("componentName", "default"));
- byte[] bytes = config.getBytes("operator", null);
-
- ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
-
- UserSourceInvokable f = (UserSourceInvokable) in.readObject();
-
- System.out.println(f.getClass().getGenericSuperclass());
- TupleTypeInfo ts = (TupleTypeInfo) TypeExtractor.createTypeInfo(
- UserSourceInvokable.class, f.getClass(), 0, null, null);
-
- System.out.println(ts);
- System.out.println("----------------");
- }
- }
+
+ fillExpectedList();
+
+ assertTrue(expected.equals(result));
+
}
}
diff --git a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/MapTest.java b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/MapTest.java
index f2a48fb3c135a0035d850f6ee6c4fff8d1764e43..1831f813056215eb1fc227b4d94c030f8b983ce8 100644
--- a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/MapTest.java
+++ b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/MapTest.java
@@ -15,123 +15,163 @@
package eu.stratosphere.streaming.api;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
-import java.io.ByteArrayInputStream;
-import java.io.ObjectInputStream;
+import java.util.ArrayList;
+import java.util.List;
+import org.jblas.util.Random;
import org.junit.Test;
-import eu.stratosphere.api.java.functions.FlatMapFunction;
import eu.stratosphere.api.java.functions.MapFunction;
-import eu.stratosphere.api.java.tuple.Tuple;
import eu.stratosphere.api.java.tuple.Tuple1;
-import eu.stratosphere.api.java.typeutils.TupleTypeInfo;
-import eu.stratosphere.api.java.typeutils.TypeExtractor;
-import eu.stratosphere.configuration.Configuration;
-import eu.stratosphere.nephele.jobgraph.AbstractJobVertex;
-import eu.stratosphere.nephele.jobgraph.JobInputVertex;
-import eu.stratosphere.nephele.jobgraph.JobOutputVertex;
-import eu.stratosphere.nephele.jobgraph.JobTaskVertex;
-import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
-import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
-import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
-import eu.stratosphere.streaming.api.streamcomponent.StreamInvokableComponent;
import eu.stratosphere.util.Collector;
public class MapTest {
- public static final class MyMap extends MapFunction, Tuple1> {
+ public static final class MySource extends SourceFunction> {
+
@Override
- public Tuple1 map(Tuple1 value) throws Exception {
- System.out.println("map");
- return value;
+ public void invoke(Collector> collector)
+ throws Exception {
+ for(int i=0; i<10; i++){
+ collector.collect(new Tuple1(i));
+ }
}
}
+
+ public static final class MyFieldsSource extends SourceFunction> {
- private static final int PARALELISM = 1;
-
- @Test
- public void test() throws Exception {
- Tuple1 tup = new Tuple1("asd");
-
- StreamExecutionEnvironment context = new StreamExecutionEnvironment();
-
- DataStream> dataStream = context.addDummySource().map(new MyMap(), PARALELISM)
- .addDummySink();
-
- context.execute();
-
- JobGraphBuilder jgb = context.jobGB();
-
- for (AbstractJobVertex c : jgb.components.values()) {
- if (c instanceof JobTaskVertex) {
- Configuration config = c.getConfiguration();
- System.out.println(config.getString("componentName", "default"));
- byte[] bytes = config.getBytes("operator", null);
-
- ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
-
- MapFunction f = (MapFunction) in.readObject();
-
- StreamCollector s = new StreamCollector(1, 1000, 1, null);
- Tuple t = new Tuple1("asd");
-
- s.collect(f.map(t));
-
- System.out.println(f.getClass().getGenericSuperclass());
- TupleTypeInfo ts = (TupleTypeInfo) TypeExtractor.createTypeInfo(
- MapFunction.class, f.getClass(), 0, null, null);
-
- System.out.println(ts);
-
- byte[] userFunctionSerialized = config.getBytes("serializedudf", null);
- in = new ObjectInputStream(new ByteArrayInputStream(userFunctionSerialized));
- UserTaskInvokable userFunction = (UserTaskInvokable) in.readObject();
- System.out.println(userFunction.getClass());
- assertTrue(true);
- System.out.println("----------------");
+ @Override
+ public void invoke(Collector> collector)
+ throws Exception {
+ for(int i=0; i(5));
}
+ }
+ }
+
+ public static final class MyMap extends MapFunction,Tuple1> {
- if (c instanceof JobOutputVertex) {
- Configuration config = c.getConfiguration();
- System.out.println(config.getString("componentName", "default"));
- byte[] bytes = config.getBytes("operator", null);
+ @Override
+ public Tuple1 map(Tuple1 value) throws Exception {
+ // TODO Auto-generated method stub
+ return new Tuple1(value.f0*value.f0);
+ }
+ }
+
+ public static final class MyFieldsMap extends MapFunction,Tuple1> {
- ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
+
+ private int counter=0;
- SinkFunction f = (SinkFunction) in.readObject();
+ @Override
+ public Tuple1 map(Tuple1 value) throws Exception {
+ // TODO Auto-generated method stub
+ counter++;
+
+ if(counter==MAXSOURCE) allInOne=true;
+ return new Tuple1(value.f0*value.f0);
+ }
+ }
+
+ public static final class MySink extends SinkFunction> {
- System.out.println(f.getClass().getGenericSuperclass());
- TupleTypeInfo ts = (TupleTypeInfo) TypeExtractor.createTypeInfo(
- SinkFunction.class, f.getClass(), 0, null, null);
+ @Override
+ public void invoke(Tuple1 tuple) {
+ result.add(tuple.f0);
+ //System.out.println("result " + tuple.f0);
+ }
+ }
+
+ public static final class MyBroadcastSink extends SinkFunction> {
- System.out.println(ts);
+ @Override
+ public void invoke(Tuple1 tuple) {
+ broadcastResult++;
+ }
+ }
+
+ public static final class MyFieldsSink extends SinkFunction> {
- byte[] userFunctionSerialized = config.getBytes("serializedudf", null);
- in = new ObjectInputStream(new ByteArrayInputStream(userFunctionSerialized));
- UserSinkInvokable userFunction = (UserSinkInvokable) in.readObject();
- System.out.println(userFunction.getClass());
- assertTrue(true);
- System.out.println("----------------");
- }
+ @Override
+ public void invoke(Tuple1 tuple) {
+ fieldsResult++;
+ }
+ }
- if (c instanceof JobInputVertex) {
- Configuration config = c.getConfiguration();
- System.out.println(config.getString("componentName", "default"));
- byte[] bytes = config.getBytes("operator", null);
+ private static List expected = new ArrayList();
+ private static List result = new ArrayList();
+ private static int broadcastResult = 0;
+ private static int fieldsResult = 0;
+ private static int fieldsCounter = 0;
+ private static final int PARALELISM = 1;
+ private static final int MAXSOURCE = 10;
+ private static boolean allInOne=false;
- ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
+ private static void fillExpectedList(){
+ for(int i=0;i<10;i++){
+ expected.add(i*i);
+ //System.out.println("expected " + i*i);
+ }
+ }
+
+ @Test
+ public void test() throws Exception {
+
+ StreamExecutionEnvironment context = new StreamExecutionEnvironment();
- UserSourceInvokable f = (UserSourceInvokable) in.readObject();
+ DataStream> dataStream = context
+ .addSource(new MySource(), 1)
+ .map(new MyMap(), PARALELISM)
+ .addSink(new MySink());
- System.out.println(f.getClass().getGenericSuperclass());
- TupleTypeInfo ts = (TupleTypeInfo) TypeExtractor.createTypeInfo(
- UserSourceInvokable.class, f.getClass(), 0, null, null);
+ context.execute();
+
+ fillExpectedList();
- System.out.println(ts);
- System.out.println("----------------");
- }
- }
+ assertTrue(expected.equals(result));
+ }
+
+ @Test
+ public void broadcastTest() throws Exception {
+ StreamExecutionEnvironment context = new StreamExecutionEnvironment();
+ DataStream> dataStream = context
+ .addSource(new MySource(), 1)
+ .broadcast()
+ .map(new MyMap(), 3)
+ .addSink(new MyBroadcastSink());
+
+ context.execute();
+ assertEquals(30, broadcastResult);
+
+ }
+
+ @Test
+ public void fieldsSinkTest() throws Exception {
+ StreamExecutionEnvironment context = new StreamExecutionEnvironment();
+ DataStream> dataStream = context
+ .addSource(new MySource(), 1)
+ .partitionBy(0)
+ .map(new MyMap(), 3)
+ .addSink(new MyFieldsSink());
+
+ context.execute();
+ assertEquals(10, fieldsResult);
+
+ }
+
+ @Test
+ public void fieldsMapTest() throws Exception {
+ StreamExecutionEnvironment context = new StreamExecutionEnvironment();
+ DataStream> dataStream = context
+ .addSource(new MyFieldsSource(), 1)
+ .partitionBy(0)
+ .map(new MyFieldsMap(), 3)
+ .addSink(new MyFieldsSink());
+
+ context.execute();
+ assertTrue(allInOne);
+
}
}
diff --git a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollector2Test.java b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollector2Test.java
index bf847fc68c44605e5fcfe22cafbeae77710f436f..3c9f0de75b0062c8421494f02d53c7f302e3a57c 100644
--- a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollector2Test.java
+++ b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollector2Test.java
@@ -15,8 +15,6 @@
package eu.stratosphere.streaming.api;
-import static org.junit.Assert.*;
-
import java.util.ArrayList;
import java.util.List;
@@ -25,20 +23,18 @@ import org.junit.Test;
import eu.stratosphere.api.java.tuple.Tuple;
import eu.stratosphere.api.java.tuple.Tuple1;
import eu.stratosphere.nephele.io.RecordWriter;
-import eu.stratosphere.streaming.api.streamcomponent.MockRecordWriter;
import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.streaming.util.MockRecordWriterFactory;
public class StreamCollector2Test {
StreamCollector2 collector;
-
+
@Test
public void testCollect() {
List batchSizesOfNotPartitioned = new ArrayList();
List batchSizesOfPartitioned = new ArrayList();
batchSizesOfPartitioned.add(2);
- batchSizesOfPartitioned.add(3);
+ batchSizesOfPartitioned.add(2);
List parallelismOfOutput = new ArrayList();
parallelismOfOutput.add(2);
parallelismOfOutput.add(2);
@@ -48,11 +44,8 @@ public class StreamCollector2Test {
List> fOut = new ArrayList>();
- MockRecordWriter rw1 = MockRecordWriterFactory.create();
- MockRecordWriter rw2 = MockRecordWriterFactory.create();
-
- fOut.add(rw1);
- fOut.add(rw2);
+ fOut.add(null);
+ fOut.add(null);
collector = new StreamCollector2(batchSizesOfNotPartitioned, batchSizesOfPartitioned, parallelismOfOutput, keyPosition, batchTimeout, channelID, null, fOut,fOut);
@@ -62,28 +55,15 @@ public class StreamCollector2Test {
t.f0 = 0;
collector.collect(t);
t.f0 = 1;
- collector.collect(t);
+ collector.collect(t);
t.f0 = 0;
collector.collect(t);
-
- StreamRecord r1 = rw1.emittedRecords.get(0);
- assertEquals(1, rw1.emittedRecords.size());
- assertEquals(0, r1.getTuple(0).getField(0));
- assertEquals(0, r1.getTuple(1).getField(0));
-
t.f0 = 1;
collector.collect(t);
-
- StreamRecord r2 = rw1.emittedRecords.get(1);
- assertEquals(2, rw1.emittedRecords.size());
- assertEquals(1, r2.getTuple(0).getField(0));
- assertEquals(1, r2.getTuple(1).getField(0));
-
- assertEquals(0, rw2.emittedRecords.size());
-
- t.f0 = 5;
- collector.collect(t);
- assertEquals(2, rw1.emittedRecords.size());
- assertEquals(1, rw2.emittedRecords.size());
}
+
+ @Test
+ public void testClose() {
+ }
+
}
diff --git a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollectorTest.java b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollectorTest.java
index 4008d6ff052e238858953b157d8156254de37d3d..c3e9ccb045738fc25250e7de50f88bf86471f7d1 100755
--- a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollectorTest.java
+++ b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/StreamCollectorTest.java
@@ -15,20 +15,11 @@
package eu.stratosphere.streaming.api;
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.doCallRealMethod;
-import static org.mockito.Mockito.mock;
-
-import java.util.ArrayList;
+import static org.junit.Assert.*;
import org.junit.Test;
-import org.mockito.Mockito;
import eu.stratosphere.api.java.tuple.Tuple1;
-import eu.stratosphere.nephele.io.RecordWriter;
-import eu.stratosphere.streaming.api.streamcomponent.MockRecordWriter;
-import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.streaming.util.MockRecordWriterFactory;
public class StreamCollectorTest {
@@ -55,30 +46,13 @@ public class StreamCollectorTest {
collector.collect(new Tuple1(0));
collector.collect(new Tuple1(0));
collector.collect(new Tuple1(0));
-
+
Thread.sleep(200);
collector.collect(new Tuple1(2));
collector.collect(new Tuple1(3));
System.out.println("---------------");
}
-
- @Test
- public void recordWriter() {
- MockRecordWriter recWriter = MockRecordWriterFactory.create();
-
- ArrayList> rwList = new ArrayList>();
- rwList.add(recWriter);
-
- StreamCollector collector = new StreamCollector(2, 1000, 0, null, rwList);
- collector.collect(new Tuple1(3));
- collector.collect(new Tuple1(4));
- collector.collect(new Tuple1(5));
- collector.collect(new Tuple1(6));
-
- assertEquals((Integer) 3, recWriter.emittedRecords.get(0).getTuple(0).getField(0));
- assertEquals((Integer) 6, recWriter.emittedRecords.get(1).getTuple(1).getField(0));
- }
-
+
@Test
public void testClose() {
}
diff --git a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/streamcomponent/MockRecordWriter.java b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/streamcomponent/MockRecordWriter.java
deleted file mode 100644
index e25d7d91b8a781341ca4e14d1142fb3e92b3f953..0000000000000000000000000000000000000000
--- a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/streamcomponent/MockRecordWriter.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/***********************************************************************************************************************
- *
- * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
- *
- * Licensed 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 eu.stratosphere.streaming.api.streamcomponent;
-
-import java.util.ArrayList;
-
-import eu.stratosphere.nephele.io.RecordWriter;
-import eu.stratosphere.nephele.template.AbstractInputTask;
-import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-
-public class MockRecordWriter extends RecordWriter {
-
- public ArrayList emittedRecords;
-
- public MockRecordWriter(AbstractInputTask> inputBase, Class outputClass) {
- super(inputBase, outputClass);
- }
-
- public boolean initList() {
- emittedRecords = new ArrayList();
- return true;
- }
-
- @Override
- public void emit(StreamRecord record) {
- emittedRecords.add(record.copy());
- }
-}
\ No newline at end of file
diff --git a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/util/MockRecordWriterFactory.java b/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/util/MockRecordWriterFactory.java
deleted file mode 100644
index 809ca978d922120bae40c5f59a870b0459ddab34..0000000000000000000000000000000000000000
--- a/flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/util/MockRecordWriterFactory.java
+++ /dev/null
@@ -1,23 +0,0 @@
-package eu.stratosphere.streaming.util;
-
-import static org.mockito.Mockito.doCallRealMethod;
-import static org.mockito.Mockito.mock;
-
-import org.mockito.Mockito;
-
-import eu.stratosphere.streaming.api.streamcomponent.MockRecordWriter;
-import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-
-public class MockRecordWriterFactory {
-
- public static MockRecordWriter create() {
- MockRecordWriter recWriter = mock(MockRecordWriter.class);
-
- Mockito.when(recWriter.initList()).thenCallRealMethod();
- doCallRealMethod().when(recWriter).emit(Mockito.any(StreamRecord.class));
-
- recWriter.initList();
-
- return recWriter;
- }
-}