flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tzuli...@apache.org
Subject [1/8] flink git commit: [hotfix] [core] Catch InvalidClassException in TypeSerializerSerializationProxy
Date Sun, 07 May 2017 19:46:59 GMT
Repository: flink
Updated Branches:
  refs/heads/master fe4e96a72 -> 63c04a516


[hotfix] [core] Catch InvalidClassException in TypeSerializerSerializationProxy

Previously, the TypeSerializerSerializationProxy only uses the dummy
ClassNotFoundDummyTypeSerializer as a placeholder in the case where the
user uses a completely new serializer and deletes the old one.

There is also the case where the user changes the original serializer's
implementation and results in an InvalidClassException when trying to
deserialize the serializer. We should also use the
ClassNotFoundDummyTypeSerializer as a temporary placeholder in this
case.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/409319a0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/409319a0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/409319a0

Branch: refs/heads/master
Commit: 409319a065407f8ed6ae3f43c06b327adfd2501c
Parents: fe4e96a
Author: Tzu-Li (Gordon) Tai <tzulitai@apache.org>
Authored: Tue May 2 19:35:18 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tzulitai@apache.org>
Committed: Mon May 8 00:33:41 2017 +0800

----------------------------------------------------------------------
 .../TypeSerializerSerializationProxy.java       |  3 +-
 .../TypeSerializerSerializationProxyTest.java   | 50 +++++++++++++++++++-
 2 files changed, 51 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/409319a0/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxy.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxy.java
b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxy.java
index cb8967b..c94124f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxy.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxy.java
@@ -29,6 +29,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.io.InvalidClassException;
 import java.util.Arrays;
 
 @Internal
@@ -97,7 +98,7 @@ public class TypeSerializerSerializationProxy<T> extends VersionedIOReadableWrit
 		in.readFully(buffer);
 		try {
 			typeSerializer = InstantiationUtil.deserializeObject(buffer, userClassLoader);
-		} catch (ClassNotFoundException e) {
+		} catch (ClassNotFoundException | InvalidClassException e) {
 			if (ignoreClassNotFound) {
 				// we create a dummy so that all the information is not lost when we get a new checkpoint
before receiving
 				// a proper typeserializer from the user

http://git-wip-us.apache.org/repos/asf/flink/blob/409319a0/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxyTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxyTest.java
b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxyTest.java
index 982e7ff..db1b4ef 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxyTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationProxyTest.java
@@ -26,13 +26,21 @@ import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
 import org.apache.flink.util.InstantiationUtil;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
 
 import java.io.IOException;
+import java.io.InvalidClassException;
 import java.net.URL;
 import java.net.URLClassLoader;
 
 import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
 
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(InstantiationUtil.class)
 public class TypeSerializerSerializationProxyTest {
 
 	@Test
@@ -91,4 +99,44 @@ public class TypeSerializerSerializationProxyTest {
 				InstantiationUtil.serializeObject(serializer),
 				((TypeSerializerSerializationProxy.ClassNotFoundDummyTypeSerializer<?>) proxy.getTypeSerializer()).getActualBytes());
 	}
-}
\ No newline at end of file
+
+	@Test
+	public void testStateSerializerSerializationProxyInvalidClass() throws Exception {
+
+		TypeSerializer<?> serializer = IntSerializer.INSTANCE;
+
+		TypeSerializerSerializationProxy<?> proxy = new TypeSerializerSerializationProxy<>(serializer);
+
+		byte[] serialized;
+		try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
+			proxy.write(new DataOutputViewStreamWrapper(out));
+			serialized = out.toByteArray();
+		}
+
+		PowerMockito.spy(InstantiationUtil.class);
+		PowerMockito
+			.doThrow(new InvalidClassException("test invalid class exception"))
+			.when(InstantiationUtil.class, "deserializeObject", any(byte[].class), any(ClassLoader.class));
+
+		proxy = new TypeSerializerSerializationProxy<>(new URLClassLoader(new URL[0], null));
+
+		try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
+			proxy.read(new DataInputViewStreamWrapper(in));
+			fail("InvalidClassException expected, leading to IOException");
+		} catch (IOException expected) {
+
+		}
+
+		proxy = new TypeSerializerSerializationProxy<>(new URLClassLoader(new URL[0], null),
true);
+
+		try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
+			proxy.read(new DataInputViewStreamWrapper(in));
+		}
+
+		Assert.assertTrue(proxy.getTypeSerializer() instanceof TypeSerializerSerializationProxy.ClassNotFoundDummyTypeSerializer);
+
+		Assert.assertArrayEquals(
+			InstantiationUtil.serializeObject(serializer),
+			((TypeSerializerSerializationProxy.ClassNotFoundDummyTypeSerializer<?>) proxy.getTypeSerializer()).getActualBytes());
+	}
+}


Mime
View raw message