flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-core/src/test/java/org/apache/flink/api/common/state/MapStateDescriptorTest.java
/**
	 * FLINK-6775
	 *
	 * Tests that the returned serializer is duplicated. This allows to
	 * share the state descriptor.
	 */
FLINK-6775
Tests that the returned serializer is duplicated. This allows to
share the state descriptor.
FLINK - 6775 Tests that the returned serializer is duplicated . This allows to share the state descriptor .
@SuppressWarnings("unchecked")
	@Test
	public void testSerializerDuplication() {
		TypeSerializer<String> keySerializer = mock(TypeSerializer.class);
		TypeSerializer<Long> valueSerializer = mock(TypeSerializer.class);
		when(keySerializer.duplicate()).thenAnswer(new Answer<TypeSerializer<String>>() {
			@Override
			public TypeSerializer<String> answer(InvocationOnMock invocation) throws Throwable {
				return mock(TypeSerializer.class);
			}
		});
		when(valueSerializer.duplicate()).thenAnswer(new Answer<TypeSerializer<Long>>() {
			@Override
			public TypeSerializer<Long> answer(InvocationOnMock invocation) throws Throwable {
				return mock(TypeSerializer.class);
			}
		});
		MapStateDescriptor<String, Long> descr = new MapStateDescriptor<>("foobar", keySerializer, valueSerializer);
		TypeSerializer<String> keySerializerA = descr.getKeySerializer();
		TypeSerializer<String> keySerializerB = descr.getKeySerializer();
		TypeSerializer<Long> valueSerializerA = descr.getValueSerializer();
		TypeSerializer<Long> valueSerializerB = descr.getValueSerializer();
		assertNotSame(keySerializerA, keySerializerB);
		assertNotSame(valueSerializerA, valueSerializerB);
		TypeSerializer<Map<String, Long>> serializerA = descr.getSerializer();
		TypeSerializer<Map<String, Long>> serializerB = descr.getSerializer();
		assertNotSame(serializerA, serializerB);
	}
@ SuppressWarnings ( "unchecked" ) @ Test public void testSerializerDuplication ( ) { TypeSerializer < String > keySerializer = mock ( TypeSerializer . class ) ; TypeSerializer < Long > valueSerializer = mock ( TypeSerializer . class ) ; when ( keySerializer . duplicate ( ) ) . thenAnswer ( new Answer < TypeSerializer < String > > ( ) { @ Override public TypeSerializer < String > answer ( InvocationOnMock invocation ) throws Throwable { return mock ( TypeSerializer . class ) ; } } ) ; when ( valueSerializer . duplicate ( ) ) . thenAnswer ( new Answer < TypeSerializer < Long > > ( ) { @ Override public TypeSerializer < Long > answer ( InvocationOnMock invocation ) throws Throwable { return mock ( TypeSerializer . class ) ; } } ) ; MapStateDescriptor < String , Long > descr = new MapStateDescriptor < > ( "foobar" , keySerializer , valueSerializer ) ; TypeSerializer < String > keySerializerA = descr . getKeySerializer ( ) ; TypeSerializer < String > keySerializerB = descr . getKeySerializer ( ) ; TypeSerializer < Long > valueSerializerA = descr . getValueSerializer ( ) ; TypeSerializer < Long > valueSerializerB = descr . getValueSerializer ( ) ; assertNotSame ( keySerializerA , keySerializerB ) ; assertNotSame ( valueSerializerA , valueSerializerB ) ; TypeSerializer < Map < String , Long > > serializerA = descr . getSerializer ( ) ; TypeSerializer < Map < String , Long > > serializerB = descr . getSerializer ( ) ; assertNotSame ( serializerA , serializerB ) ; }
FLINK-6775
 Tests that the returned serializer is duplicated.
FLINK - 6775 Tests that the returned serializer is duplicated .
// check that we did not retrieve the same serializers