[streaming] Implemented AckEvent and FailEvent serialization

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

@@ -1149,7 +1149,6 @@ public class StreamRecord implements IOReadableWritable, Serializable {
 	Class[] tupleBasicTypesFromString(String representation, int numberOfFields) {
 		// TODO: use StringTokenizer
 		String[] types = representation.split(",");
-		@SuppressWarnings("rawtypes")
 		Class[] basicTypes = new Class[types.length];
 		for (int i = 0; i < types.length; i++) {
 			try {

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/faulttolerance/AckEvent.java

@@ -20,6 +20,7 @@ import java.io.DataOutput;
 import java.io.IOException;
 
 import eu.stratosphere.nephele.event.task.AbstractTaskEvent;
+import eu.stratosphere.types.StringValue;
 
 /**
  * TaskEvent for sending record acknowledgements to the input's fault tolerance
@@ -41,10 +42,15 @@ public class AckEvent extends AbstractTaskEvent {
 
 	@Override
 	public void write(DataOutput out) throws IOException {
+		StringValue recordIdValue = new StringValue(recordId);
+		recordIdValue.write(out);
 	}
 
 	@Override
 	public void read(DataInput in) throws IOException {
+		StringValue recordIdValue = new StringValue("");
+		recordIdValue.read(in);
+		setRecordId(recordIdValue.getValue());
 	}
 
 	public void setRecordId(String recordId) {

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/faulttolerance/FailEvent.java

@@ -20,6 +20,7 @@ import java.io.DataOutput;
 import java.io.IOException;
 
 import eu.stratosphere.nephele.event.task.AbstractTaskEvent;
+import eu.stratosphere.types.StringValue;
 
 /**
  * TaskEvent for sending record fails to the input's fault tolerance buffer
@@ -42,10 +43,15 @@ public class FailEvent extends AbstractTaskEvent {
 
 	@Override
 	public void write(DataOutput out) throws IOException {
+		StringValue recordIdValue = new StringValue(recordId);
+		recordIdValue.write(out);
 	}
 
 	@Override
 	public void read(DataInput in) throws IOException {
+		StringValue recordIdValue = new StringValue("");
+		recordIdValue.read(in);
+		setRecordId(recordIdValue.getValue());
 	}
 
 	public void setRecordId(String recordId) {