[FLINK-1371] [runtime] Fix KryoSerializer to not swallow EOFExceptions

flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java

@@ -30,6 +30,7 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
 
 public class KryoSerializer<T> extends TypeSerializer<T> {
@@ -114,8 +115,19 @@ public void serialize(T record, DataOutputView target) throws IOException {
 			previousOut = target;
 		}
 
-		kryo.writeClassAndObject(output, record);
-		output.flush();
+		try {
+			kryo.writeClassAndObject(output, record);
+			output.flush();
+		}
+		catch (KryoException ke) {
+			Throwable cause = ke.getCause();
+			if (cause instanceof EOFException) {
+				throw (EOFException) cause;
+			}
+			else {
+				throw ke;
+			}
+		}
 	}
 
 	@SuppressWarnings("unchecked")
@@ -173,4 +185,13 @@ private void checkKryoInitialized() {
 			this.kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
 		}
 	}
+
+	// --------------------------------------------------------------------------------------------
+	// for testing
+	// --------------------------------------------------------------------------------------------
+
+	Kryo getKryo() {
+		checkKryoInitialized();
+		return this.kryo;
+	}
 }

flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java

@@ -43,6 +43,7 @@ public WritableSerializer(Class<T> typeClass) {
 		this.typeClass = typeClass;
 	}
 
+	@SuppressWarnings("unchecked")
 	@Override
 	public T createInstance() {
 		if(typeClass == NullWritable.class) {

flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java

@@ -18,17 +18,20 @@
 
 package org.apache.flink.api.java.typeutils.runtime;
 
+import static org.junit.Assert.*;
 
-import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.junit.Test;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.LinkedList;
+import java.util.Random;
 
+@SuppressWarnings("unchecked")
 public class KryoGenericTypeSerializerTest extends AbstractGenericTypeSerializerTest {
-
+	
 	@Test
 	public void testJavaList(){
 		Collection<Integer> a = new ArrayList<Integer>();
@@ -67,4 +70,44 @@ private void fillCollection(Collection<Integer> coll){
 	protected <T> TypeSerializer<T> createSerializer(Class<T> type) {
 		return new KryoSerializer<T>(type);
 	}
+
+	/**
+	 * Make sure that the kryo serializer forwards EOF exceptions properly
+	 */
+	@Test
+	public void testForwardEOFException() {
+		try {
+			// construct a long string
+			String str;
+			{
+				char[] charData = new char[40000];
+				Random rnd = new Random();
+
+				for (int i = 0; i < charData.length; i++) {
+					charData[i] = (char) rnd.nextInt(10000);
+				}
+
+				str = new String(charData);
+			}
+
+			// construct a memory target that is too small for the string
+			TestDataOutputSerializer target = new TestDataOutputSerializer(10000, 30000);
+			KryoSerializer<String> serializer = new KryoSerializer<String>(String.class);
+
+			try {
+				serializer.serialize(str, target);
+				fail("should throw a java.io.EOFException");
+			}
+			catch (java.io.EOFException e) {
+				// that is how we like it
+			}
+			catch (Exception e) {
+				fail("throws wrong exception: should throw a java.io.EOFException, has thrown a " + e.getClass().getName());
+			}
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
 }