[streaming] api cleanup

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

@@ -69,6 +69,8 @@ public class JobGraphBuilder {
 	 * 
 	 * @param jobGraphName
 	 *            Name of the JobGraph
+	 * @param faultToleranceType
+	 *            Fault tolerance type
 	 */
 	public JobGraphBuilder(String jobGraphName, FaultToleranceType faultToleranceType) {
 		jobGraph = new JobGraph(jobGraphName);
@@ -81,6 +83,17 @@ public class JobGraphBuilder {
 		this.faultToleranceType = faultToleranceType;
 	}
 
+	/**
+	 * Creates a new JobGraph with the given name with fault tolerance turned
+	 * off
+	 * 
+	 * @param jobGraphName
+	 *            Name of the JobGraph
+	 */
+	public JobGraphBuilder(String jobGraphName) {
+		this(jobGraphName, FaultToleranceType.NONE);
+	}
+
 	/**
 	 * Adds a source component to the JobGraph with no parallelism
 	 * 

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

@@ -52,6 +52,10 @@ public abstract class StreamInvokableComponent implements Serializable {
 		this.useFaultTolerance = faultToleranceType != FaultToleranceType.NONE;
 	}
 
+	public final void setPerfCounterDir(String dir) {
+		performanceCounter.setFname(dir + "/" + name + channelID);
+	}
+
 	public final void emit(StreamRecord record) {
 		record.setId(channelID);
 
@@ -69,7 +73,7 @@ public abstract class StreamInvokableComponent implements Serializable {
 			if (useFaultTolerance) {
 				emittedRecords.failRecord(record.getId());
 			}
-			
+
 			log.warn("FAILED: " + record.getId() + " -- " + name + " -- due to "
 					+ e.getClass().getSimpleName());
 		}

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

@@ -1324,8 +1324,12 @@ public class StreamRecord implements IOReadableWritable, Serializable {
 	public String toString() {
 		StringBuilder outputString = new StringBuilder("[");
 
+		String prefix = "";
+
 		for (Tuple tuple : tupleBatch) {
-			outputString.append(tuple + ",");
+			outputString.append(prefix);
+			prefix = ",";
+			outputString.append(tuple.toString());
 		}
 		outputString.append("]");
 		return outputString.toString();

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/basictopology/BasicTopology.java

@@ -14,14 +14,9 @@
  **********************************************************************************************************************/
 package eu.stratosphere.streaming.examples.basictopology;
 
-import java.net.InetSocketAddress;
-
 import org.apache.log4j.Level;
 
 import eu.stratosphere.api.java.tuple.Tuple1;
-import eu.stratosphere.client.minicluster.NepheleMiniCluster;
-import eu.stratosphere.client.program.Client;
-import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
@@ -29,12 +24,14 @@ import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
 import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
 import eu.stratosphere.streaming.faulttolerance.FaultToleranceType;
+import eu.stratosphere.streaming.util.ClusterUtil;
 import eu.stratosphere.streaming.util.LogUtils;
 
 public class BasicTopology {
 
 	public static class BasicSource extends UserSourceInvokable {
 
+		private static final long serialVersionUID = 1L;
 		StreamRecord record = new StreamRecord(new Tuple1<String>("streaming"));
 
 		@Override
@@ -50,6 +47,7 @@ public class BasicTopology {
 	}
 
 	public static class BasicTask extends UserTaskInvokable {
+		private static final long serialVersionUID = 1L;
 
 		@Override
 		public void invoke(StreamRecord record) throws Exception {
@@ -61,19 +59,20 @@ public class BasicTopology {
 	}
 
 	public static class BasicSink extends UserSinkInvokable {
+		private static final long serialVersionUID = 1L;
 
 		@Override
 		public void invoke(StreamRecord record) throws Exception {
 			// do nothing
-			record.getField(0);
+			System.out.println(record.getField(0));
 		}
 	}
 
-	private static JobGraph getJobGraph() throws Exception {
-		JobGraphBuilder graphBuilder = new JobGraphBuilder("BasicStreamingTopology", FaultToleranceType.NONE);
-		graphBuilder.setSource("BasicSource", BasicSource.class, 1, 1);
-		graphBuilder.setTask("BasicTask", BasicTask.class, 1, 1);
-		graphBuilder.setSink("BasicSink", BasicSink.class, 1, 1);
+	private static JobGraph getJobGraph() {
+		JobGraphBuilder graphBuilder = new JobGraphBuilder("BasicStreamingTopology");
+		graphBuilder.setSource("BasicSource", new BasicSource(), 1, 1);
+		graphBuilder.setTask("BasicTask", new BasicTask(), 1, 1);
+		graphBuilder.setSink("BasicSink", new BasicSink(), 1, 1);
 
 		graphBuilder.shuffleConnect("BasicSource", "BasicTask");
 		graphBuilder.shuffleConnect("BasicTask", "BasicSink");
@@ -84,34 +83,8 @@ public class BasicTopology {
 	public static void main(String[] args) {
 
 		// set logging parameters for local run
-		LogUtils.initializeDefaultConsoleLogger(Level.ERROR, Level.INFO);
-
-		try {
-
-			// generate JobGraph
-			JobGraph jG = getJobGraph();
-			Configuration configuration = jG.getJobConfiguration();
-
-			if (args.length == 0 || args[0].equals("local")) {
-				System.out.println("Running in Local mode");
-				// start local cluster and submit JobGraph
-				NepheleMiniCluster exec = new NepheleMiniCluster();
-				exec.start();
-
-				Client client = new Client(new InetSocketAddress("localhost", 6498), configuration);
+		LogUtils.initializeDefaultConsoleLogger(Level.INFO, Level.INFO);
 
-				client.run(jG, true);
-
-				exec.stop();
-			} else if (args[0].equals("cluster")) {
-				System.out.println("Running in Cluster mode");
-				// submit JobGraph to the running cluster
-				Client client = new Client(new InetSocketAddress("dell150", 6123), configuration);
-				client.run(jG, true);
-			}
-
-		} catch (Exception e) {
-			System.out.println(e);
-		}
+		ClusterUtil.runOnMiniCluster(getJobGraph());
 	}
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/ml/IncrementalLearningSkeleton.java

@@ -154,11 +154,11 @@ public class IncrementalLearningSkeleton {
 		JobGraphBuilder graphBuilder = new JobGraphBuilder("IncrementalLearning",
 				FaultToleranceType.NONE);
 
-		graphBuilder.setSource("NewData", NewDataSource.class, 1, 1);
-		graphBuilder.setSource("TrainingData", TrainingDataSource.class, 1, 1);
-		graphBuilder.setTask("PartialModelBuilder", PartialModelBuilder.class, 1, 1);
-		graphBuilder.setTask("Predictor", Predictor.class, 1, 1);
-		graphBuilder.setSink("Sink", Sink.class, 1, 1);
+		graphBuilder.setSource("NewData", new NewDataSource(), 1, 1);
+		graphBuilder.setSource("TrainingData",new TrainingDataSource(), 1, 1);
+		graphBuilder.setTask("PartialModelBuilder",new PartialModelBuilder(), 1, 1);
+		graphBuilder.setTask("Predictor",new Predictor(), 1, 1);
+		graphBuilder.setSink("Sink",new Sink(), 1, 1);
 
 		graphBuilder.shuffleConnect("TrainingData", "PartialModelBuilder");
 		graphBuilder.shuffleConnect("NewData", "Predictor");

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/ml/IncrementalOLS.java

@@ -180,11 +180,11 @@ public class IncrementalOLS {
 		JobGraphBuilder graphBuilder = new JobGraphBuilder("IncrementalOLS",
 				FaultToleranceType.NONE);
 
-		graphBuilder.setSource("NewData", NewDataSource.class, 1, 1);
-		graphBuilder.setSource("TrainingData", TrainingDataSource.class, 1, 1);
-		graphBuilder.setTask("PartialModelBuilder", PartialModelBuilder.class, 1, 1);
-		graphBuilder.setTask("Predictor", Predictor.class, 1, 1);
-		graphBuilder.setSink("Sink", Sink.class, 1, 1);
+		graphBuilder.setSource("NewData", new NewDataSource(), 1, 1);
+		graphBuilder.setSource("TrainingData",new TrainingDataSource(), 1, 1);
+		graphBuilder.setTask("PartialModelBuilder",new PartialModelBuilder(), 1, 1);
+		graphBuilder.setTask("Predictor",new Predictor(), 1, 1);
+		graphBuilder.setSink("Sink",new Sink(), 1, 1);
 
 		graphBuilder.shuffleConnect("TrainingData", "PartialModelBuilder");
 		graphBuilder.shuffleConnect("NewData", "Predictor");

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/wordcount/WordCountCounter.java

@@ -21,6 +21,7 @@ import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
 import eu.stratosphere.streaming.state.MutableTableState;
 
 public class WordCountCounter extends UserTaskInvokable {
+	private static final long serialVersionUID = 1L;
 
 	private MutableTableState<String, Integer> wordCounts = new MutableTableState<String, Integer>();
 	private String word = "";

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/wordcount/WordCountRemote.java

@@ -42,6 +42,7 @@ public class WordCountRemote {
 	private final static int recordsEmitted = 100000;
 
 	public static class WordCountDebugSource extends UserSourceInvokable {
+		private static final long serialVersionUID = 1L;
 
 		private PerformanceCounter perf = new PerformanceCounter("SourceEmitCounter", 1000, 10000, "");
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/wordcount/WordCountSink.java

@@ -19,6 +19,7 @@ import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
 
 public class WordCountSink extends UserSinkInvokable {
+	private static final long serialVersionUID = 1L;
 
 	@Override
 	public void invoke(StreamRecord record) throws Exception {

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/wordcount/WordCountSource.java

@@ -23,6 +23,7 @@ import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
 
 public class WordCountSource extends UserSourceInvokable {
+	private static final long serialVersionUID = 1L;
 
 	private BufferedReader br = null;
 	private String line = new String();

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/wordcount/WordCountSplitter.java

@@ -22,6 +22,7 @@ import eu.stratosphere.streaming.util.PerformanceCounter;
 import eu.stratosphere.streaming.util.PerformanceTimer;
 
 public class WordCountSplitter extends UserTaskInvokable {
+	private static final long serialVersionUID = 1L;
 
 	private String[] words = new String[] {};
 	private StreamRecord outputRecord = new StreamRecord(new Tuple1<String>());

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/rabbitmq/RMQSource.java

@@ -87,7 +87,7 @@ public class RMQSource extends UserSourceInvokable {
 
 			message = new String(delivery.getBody());
 
-			if (message.equals("quit")) {
+			if (message.equals("q")) {
 				break;
 			}
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/rabbitmq/RMQTopology.java

@@ -27,7 +27,6 @@ import eu.stratosphere.streaming.util.LogUtils;
 
 public class RMQTopology {
 
-
 	public static class Sink extends UserSinkInvokable {
 		private static final long serialVersionUID = 1L;
 
@@ -41,7 +40,7 @@ public class RMQTopology {
 
 		JobGraphBuilder graphBuilder = new JobGraphBuilder("RMQ", FaultToleranceType.NONE);
 		graphBuilder.setSource("Source", new RMQSource("localhost", "hello"), 1, 1);
-		graphBuilder.setSink("Sink", Sink.class, 1, 1);
+		graphBuilder.setSink("Sink", new Sink(), 1, 1);
 
 		graphBuilder.shuffleConnect("Source", "Sink");
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/util/PerformanceTracker.java

@@ -148,4 +148,8 @@ public class PerformanceTracker {
 
 	}
 
+	public void setFname(String fname) {
+		this.fname = fname;
+	}
+
 }

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

@@ -17,7 +17,6 @@ package eu.stratosphere.streaming.api.streamcomponent;
 
 import static org.junit.Assert.assertEquals;
 
-import java.net.InetSocketAddress;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -27,10 +26,6 @@ import org.junit.Test;
 
 import eu.stratosphere.api.java.tuple.Tuple1;
 import eu.stratosphere.api.java.tuple.Tuple2;
-import eu.stratosphere.client.minicluster.NepheleMiniCluster;
-import eu.stratosphere.client.program.Client;
-import eu.stratosphere.configuration.Configuration;
-import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
 import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;