flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationUtilTest.java
/**
	 * Verifies resilience to serializer deserialization failures when writing and reading
	 * serializer and config snapshot pairs.
	 */
Verifies resilience to serializer deserialization failures when writing and reading
serializer and config snapshot pairs.
Verifies resilience to serializer deserialization failures when writing and reading serializer and config snapshot pairs .
@Test
	public void testSerializerAndConfigPairsSerializationWithSerializerDeserializationFailures() throws Exception {
		List<Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot>> serializersAndConfigs = Arrays.asList(
			new Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot>(
				IntSerializer.INSTANCE, IntSerializer.INSTANCE.snapshotConfiguration()),
			new Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot>(
				DoubleSerializer.INSTANCE, DoubleSerializer.INSTANCE.snapshotConfiguration()));
		byte[] serializedSerializersAndConfigs;
		try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
			TypeSerializerSerializationUtil.writeSerializersAndConfigsWithResilience(
					new DataOutputViewStreamWrapper(out), serializersAndConfigs);
			serializedSerializersAndConfigs = out.toByteArray();
		}
		TypeSerializerSerializationUtil.TypeSerializerSerializationProxy<?> mockProxy =
				mock(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class);
		doThrow(new IOException()).when(mockProxy).read(any(DataInputViewStreamWrapper.class));
		PowerMockito.whenNew(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class).withAnyArguments().thenReturn(mockProxy);
		List<Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot>> restored;
		try (ByteArrayInputStream in = new ByteArrayInputStream(serializedSerializersAndConfigs)) {
			restored = TypeSerializerSerializationUtil.readSerializersAndConfigsWithResilience(
				new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
		}
		Assert.assertEquals(2, restored.size());
		Assert.assertEquals(null, restored.get(0).f0);
		Assert.assertEquals(IntSerializer.INSTANCE.snapshotConfiguration(), restored.get(0).f1);
		Assert.assertEquals(null, restored.get(1).f0);
		Assert.assertEquals(DoubleSerializer.INSTANCE.snapshotConfiguration(), restored.get(1).f1);
	}
@ Test public void testSerializerAndConfigPairsSerializationWithSerializerDeserializationFailures ( ) throws Exception { List < Tuple2 < TypeSerializer < ? > , TypeSerializerConfigSnapshot > > serializersAndConfigs = Arrays . asList ( new Tuple2 < TypeSerializer < ? > , TypeSerializerConfigSnapshot > ( IntSerializer . INSTANCE , IntSerializer . INSTANCE . snapshotConfiguration ( ) ) , new Tuple2 < TypeSerializer < ? > , TypeSerializerConfigSnapshot > ( DoubleSerializer . INSTANCE , DoubleSerializer . INSTANCE . snapshotConfiguration ( ) ) ) ; byte [ ] serializedSerializersAndConfigs ; try ( ByteArrayOutputStream out = new ByteArrayOutputStream ( ) ) { TypeSerializerSerializationUtil . writeSerializersAndConfigsWithResilience ( new DataOutputViewStreamWrapper ( out ) , serializersAndConfigs ) ; serializedSerializersAndConfigs = out . toByteArray ( ) ; } TypeSerializerSerializationUtil . TypeSerializerSerializationProxy < ? > mockProxy = mock ( TypeSerializerSerializationUtil . TypeSerializerSerializationProxy . class ) ; doThrow ( new IOException ( ) ) . when ( mockProxy ) . read ( any ( DataInputViewStreamWrapper . class ) ) ; PowerMockito . whenNew ( TypeSerializerSerializationUtil . TypeSerializerSerializationProxy . class ) . withAnyArguments ( ) . thenReturn ( mockProxy ) ; List < Tuple2 < TypeSerializer < ? > , TypeSerializerConfigSnapshot > > restored ; try ( ByteArrayInputStream in = new ByteArrayInputStream ( serializedSerializersAndConfigs ) ) { restored = TypeSerializerSerializationUtil . readSerializersAndConfigsWithResilience ( new DataInputViewStreamWrapper ( in ) , Thread . currentThread ( ) . getContextClassLoader ( ) ) ; } Assert . assertEquals ( 2 , restored . size ( ) ) ; Assert . assertEquals ( null , restored . get ( 0 ) . f0 ) ; Assert . assertEquals ( IntSerializer . INSTANCE . snapshotConfiguration ( ) , restored . get ( 0 ) . f1 ) ; Assert . assertEquals ( null , restored . get ( 1 ) . f0 ) ; Assert . assertEquals ( DoubleSerializer . INSTANCE . snapshotConfiguration ( ) , restored . get ( 1 ) . f1 ) ; }
Verifies resilience to serializer deserialization failures when writing and reading
 serializer and config snapshot pairs.
Verifies resilience to serializer deserialization failures when writing and reading serializer and config snapshot pairs .
// mock failure when deserializing serializers