[streaming] updated fields partitioning api

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

@@ -236,8 +236,7 @@ public class JobGraphBuilder {
 	 *            Class of the key Value stored in the record
 	 */
 	//TODO: remove unused third parameter
-	public void fieldsConnect(String upStreamComponentName, String downStreamComponentName, int keyPosition,
-			Class<? extends Key> keyClass) {
+	public void fieldsConnect(String upStreamComponentName, String downStreamComponentName, int keyPosition) {
 
 		AbstractJobVertex upStreamComponent = components.get(upStreamComponentName);
 		AbstractJobVertex downStreamComponent = components.get(downStreamComponentName);
@@ -250,9 +249,6 @@ public class JobGraphBuilder {
 			config.setClass("partitionerClass_" + (upStreamComponent.getNumberOfForwardConnections() - 1),
 					FieldsPartitioner.class);
 
-			config.setClass("partitionerClassParam_" + (upStreamComponent.getNumberOfForwardConnections() - 1),
-					keyClass);
-
 			config.setInteger("partitionerIntParam_" + (upStreamComponent.getNumberOfForwardConnections() - 1),
 					keyPosition);
 

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

@@ -42,8 +42,6 @@ import eu.stratosphere.streaming.faulttolerance.FailEventListener;
 import eu.stratosphere.streaming.faulttolerance.FaultToleranceUtil;
 import eu.stratosphere.streaming.partitioner.DefaultPartitioner;
 import eu.stratosphere.streaming.partitioner.FieldsPartitioner;
-import eu.stratosphere.types.Key;
-import eu.stratosphere.types.StringValue;
 
 public final class StreamComponentHelper<T extends AbstractInvokable> {
 	private static final Log log = LogFactory.getLog(StreamComponentHelper.class);
@@ -171,11 +169,10 @@ public final class StreamComponentHelper<T extends AbstractInvokable> {
 			if (partitioner.equals(FieldsPartitioner.class)) {
 				int keyPosition = taskConfiguration
 						.getInteger("partitionerIntParam_" + nrOutput, 1);
-				Class<? extends Key> keyClass = taskConfiguration.getClass("partitionerClassParam_"
-						+ nrOutput, StringValue.class, Key.class);
 				
-				partitioners.add(partitioner.getConstructor(int.class, Class.class).newInstance(
-						keyPosition, keyClass));
+
+				partitioners.add(partitioner.getConstructor(int.class).newInstance(
+						keyPosition));
 
 			} else {
 				partitioners.add(partitioner.newInstance());

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

@@ -25,7 +25,6 @@ import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.streaming.util.LogUtils;
-import eu.stratosphere.types.StringValue;
 
 public class BatchWordCountLocal {
 
@@ -37,8 +36,7 @@ public class BatchWordCountLocal {
 		graphBuilder.setSink("BatchWordCountSink", BatchWordCountSink.class);
 
 		graphBuilder.shuffleConnect("BatchWordCountSource", "BatchWordCountSplitter");
-		graphBuilder.fieldsConnect("BatchWordCountSplitter", "BatchWordCountCounter", 0,
-				StringValue.class);
+		graphBuilder.fieldsConnect("BatchWordCountSplitter", "BatchWordCountCounter", 0);
 		graphBuilder.shuffleConnect("BatchWordCountCounter", "BatchWordCountSink");
 
 		return graphBuilder.getJobGraph();

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/cellinfo/CellInfoLocal.java

@@ -25,7 +25,6 @@ import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.streaming.util.LogUtils;
-import eu.stratosphere.types.IntValue;
 
 public class CellInfoLocal {
 
@@ -36,9 +35,8 @@ public class CellInfoLocal {
 		graphBuilder.setTask("cellTask", CellTask.class, 3);
 		graphBuilder.setSink("sink", CellSink.class);
 
-		graphBuilder.fieldsConnect("infoSource", "cellTask", 0, IntValue.class);
-		graphBuilder
-				.fieldsConnect("querySource", "cellTask", 0, IntValue.class);
+		graphBuilder.fieldsConnect("infoSource", "cellTask", 0);
+		graphBuilder.fieldsConnect("querySource", "cellTask", 0);
 		graphBuilder.shuffleConnect("cellTask", "sink");
 
 		return graphBuilder.getJobGraph();
@@ -63,8 +61,7 @@ public class CellInfoLocal {
 
 				exec.start();
 
-				Client client = new Client(new InetSocketAddress("localhost",
-						6498), configuration);
+				Client client = new Client(new InetSocketAddress("localhost", 6498), configuration);
 
 				client.run(jG, true);
 
@@ -72,8 +69,8 @@ public class CellInfoLocal {
 			} else if (args[0].equals("cluster")) {
 				System.out.println("Running in Cluster2 mode");
 
-				Client client = new Client(new InetSocketAddress(
-						"hadoop02.ilab.sztaki.hu", 6123), configuration);
+				Client client = new Client(new InetSocketAddress("hadoop02.ilab.sztaki.hu", 6123),
+						configuration);
 				client.run(jG, true);
 			}
 

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

@@ -25,7 +25,6 @@ import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.streaming.util.LogUtils;
-import eu.stratosphere.types.StringValue;
 
 //TODO: window operator remains unfinished.
 public class WindowWordCountLocal {
@@ -38,8 +37,7 @@ public class WindowWordCountLocal {
 		graphBuilder.setSink("WindowWordCountSink", WindowWordCountSink.class);
 
 		graphBuilder.broadcastConnect("WindowWordCountSource", "WindowWordCountSplitter");
-		graphBuilder.fieldsConnect("WindowWordCountSplitter", "WindowWordCountCounter", 0,
-				StringValue.class);
+		graphBuilder.fieldsConnect("WindowWordCountSplitter", "WindowWordCountCounter", 0);
 		graphBuilder.broadcastConnect("WindowWordCountCounter", "WindowWordCountSink");
 
 		return graphBuilder.getJobGraph();

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

@@ -25,7 +25,6 @@ import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.streaming.util.LogUtils;
-import eu.stratosphere.types.StringValue;
 
 public class WordCountLocal {
 
@@ -37,7 +36,7 @@ public class WordCountLocal {
 		graphBuilder.setSink("WordCountSink", WordCountSink.class);
 
 		graphBuilder.shuffleConnect("WordCountSource", "WordCountSplitter");
-		graphBuilder.fieldsConnect("WordCountSplitter", "WordCountCounter", 0, StringValue.class);
+		graphBuilder.fieldsConnect("WordCountSplitter", "WordCountCounter", 0);
 		graphBuilder.shuffleConnect("WordCountCounter", "WordCountSink");
 				
 		return graphBuilder.getJobGraph();

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

@@ -27,7 +27,6 @@ import eu.stratosphere.core.fs.Path;
 import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.streaming.util.LogUtils;
-import eu.stratosphere.types.StringValue;
 
 public class WordCountRemote {
 
@@ -39,7 +38,7 @@ public class WordCountRemote {
 		graphBuilder.setSink("WordCountSink", WordCountSink.class);
 
 		graphBuilder.shuffleConnect("WordCountSource", "WordCountSplitter");
-		graphBuilder.fieldsConnect("WordCountSplitter", "WordCountCounter", 0, StringValue.class);
+		graphBuilder.fieldsConnect("WordCountSplitter", "WordCountCounter", 0);
 		graphBuilder.shuffleConnect("WordCountCounter", "WordCountSink");
 
 		return graphBuilder.getJobGraph();

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/partitioner/FieldsPartitioner.java

@@ -17,35 +17,20 @@ package eu.stratosphere.streaming.partitioner;
 
 import eu.stratosphere.nephele.io.ChannelSelector;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.types.Key;
 
 //Grouping by a key
 public class FieldsPartitioner implements ChannelSelector<StreamRecord> {
 
 	private int keyPosition;
-//	private Class<? extends Key> keyClass;
 
-	public FieldsPartitioner(int keyPosition, Class<? extends Key> keyClass) {
+	public FieldsPartitioner(int keyPosition) {
 		this.keyPosition = keyPosition;
-//		this.keyClass = keyClass;
 	}
 
 	@Override
 	public int[] selectChannels(StreamRecord record, int numberOfOutputChannels) {
-		//TODO:fix this
-//		Key key = null;
-//		try {
-//			key = keyClass.newInstance();
-//		} catch (InstantiationException e) {
-//			e.printStackTrace();
-//		} catch (IllegalAccessException e) {
-//			e.printStackTrace();
-//		}
-		// TODO: consider hash partition the whole record batch.
+		//TODO:Better hashing?
 
-//
-//		}
-//		key = keyClass.cast(record.getField(0, keyPosition));
 		return new int[] { Math.abs(record.getField(0, keyPosition).hashCode()) % numberOfOutputChannels };
 	}
 }

flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/faulttolerance/AtLeastOnceBufferTest.java

@@ -15,14 +15,16 @@
 
 package eu.stratosphere.streaming.faulttolerance;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 import org.junit.Before;
 import org.junit.Test;
 
 import eu.stratosphere.api.java.tuple.Tuple1;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.types.StringValue;
 
 public class AtLeastOnceBufferTest {
 

flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/faulttolerance/ExactlyOnceBufferTest.java

@@ -15,14 +15,16 @@
 
 package eu.stratosphere.streaming.faulttolerance;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 import org.junit.Before;
 import org.junit.Test;
 
 import eu.stratosphere.api.java.tuple.Tuple1;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.types.StringValue;
 
 public class ExactlyOnceBufferTest {
 

flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/faulttolerance/FaultToleranceUtilTest.java

@@ -15,12 +15,6 @@
 
 package eu.stratosphere.streaming.faulttolerance;
 
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 
@@ -29,8 +23,6 @@ import org.junit.Test;
 
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.streaming.faulttolerance.FaultToleranceUtil;
-import eu.stratosphere.types.StringValue;
 
 public class FaultToleranceUtilTest {
 

flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/partitioner/FieldsPartitionerTest.java

@@ -23,7 +23,6 @@ import org.junit.Test;
 
 import eu.stratosphere.api.java.tuple.Tuple2;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.types.StringValue;
 
 public class FieldsPartitionerTest {
 
@@ -35,7 +34,7 @@ public class FieldsPartitionerTest {
 
 	@Before
 	public void setPartitioner() {
-		fieldsPartitioner = new FieldsPartitioner(0, StringValue.class);
+		fieldsPartitioner = new FieldsPartitioner(0);
 	}
 
 	@Test