ListStateDescriptor extends StateDescriptor<ListState<T>, List<T>> { public TypeSerializer<T> getElementSerializer() { final TypeSerializer<List<T>> rawSerializer = getSerializer(); if (!(rawSerializer instanceof ListSerializer)) { throw new IllegalStateException(); } return ((ListSerializer<T>) rawSerializer).getElementSerializer(); } ListStateDescriptor(String name, Class<T> elementTypeClass);  ListStateDescriptor(String name, TypeInformation<T> elementTypeInfo);  ListStateDescriptor(String name, TypeSerializer<T> typeSerializer); @Override ListState<T> bind(StateBinder stateBinder); TypeSerializer<T> getElementSerializer(); @Override Type getType(); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString();  }
@Test public void testValueStateDescriptorEagerSerializer() throws Exception { TypeSerializer<String> serializer = new KryoSerializer<>(String.class, new ExecutionConfig()); ListStateDescriptor<String> descr = new ListStateDescriptor<String>("testName", serializer); assertEquals("testName", descr.getName()); assertNotNull(descr.getSerializer()); assertTrue(descr.getSerializer() instanceof ListSerializer); assertNotNull(descr.getElementSerializer()); assertEquals(serializer, descr.getElementSerializer()); ListStateDescriptor<String> copy = CommonTestUtils.createCopySerializable(descr); assertEquals("testName", copy.getName()); assertNotNull(copy.getSerializer()); assertTrue(copy.getSerializer() instanceof ListSerializer); assertNotNull(copy.getElementSerializer()); assertEquals(serializer, copy.getElementSerializer()); }
@Test public void testValueStateDescriptorLazySerializer() throws Exception { ExecutionConfig cfg = new ExecutionConfig(); cfg.registerKryoType(TaskInfo.class); ListStateDescriptor<Path> descr = new ListStateDescriptor<Path>("testName", Path.class); try { descr.getSerializer(); fail("should cause an exception"); } catch (IllegalStateException ignored) {} descr.initializeSerializerUnlessSet(cfg); assertNotNull(descr.getSerializer()); assertTrue(descr.getSerializer() instanceof ListSerializer); assertNotNull(descr.getElementSerializer()); assertTrue(descr.getElementSerializer() instanceof KryoSerializer); assertTrue(((KryoSerializer<?>) descr.getElementSerializer()).getKryo().getRegistration(TaskInfo.class).getId() > 0); }
@Test public void testValueStateDescriptorAutoSerializer() throws Exception { ListStateDescriptor<String> descr = new ListStateDescriptor<String>("testName", String.class); ListStateDescriptor<String> copy = CommonTestUtils.createCopySerializable(descr); assertEquals("testName", copy.getName()); assertNotNull(copy.getSerializer()); assertTrue(copy.getSerializer() instanceof ListSerializer); assertNotNull(copy.getElementSerializer()); assertEquals(StringSerializer.INSTANCE, copy.getElementSerializer()); }
@SuppressWarnings("unchecked") @Test public void testSerializerDuplication() { TypeSerializer<String> statefulSerializer = mock(TypeSerializer.class); when(statefulSerializer.duplicate()).thenAnswer(new Answer<TypeSerializer<String>>() { @Override public TypeSerializer<String> answer(InvocationOnMock invocation) throws Throwable { return mock(TypeSerializer.class); } }); ListStateDescriptor<String> descr = new ListStateDescriptor<>("foobar", statefulSerializer); TypeSerializer<String> serializerA = descr.getElementSerializer(); TypeSerializer<String> serializerB = descr.getElementSerializer(); assertNotSame(serializerA, serializerB); TypeSerializer<List<String>> listSerializerA = descr.getSerializer(); TypeSerializer<List<String>> listSerializerB = descr.getSerializer(); assertNotSame(listSerializerA, listSerializerB); }