@Override
    @SuppressWarnings("unchecked")
    public TypeSerializer<T> restoreSerializer() {
        final int numFields = snapshotData.getFieldSerializerSnapshots().size();

        final ArrayList<Field> restoredFields = new ArrayList<>(numFields);
        final ArrayList<TypeSerializer<?>> restoredFieldSerializers = new ArrayList<>(numFields);
        snapshotData
                .getFieldSerializerSnapshots()
                .forEach(
                        (fieldName, field, fieldSerializerSnapshot) -> {
                            restoredFields.add(field);
                            checkState(
                                    fieldSerializerSnapshot != null,
                                    "field serializer snapshots should be present.");
                            restoredFieldSerializers.add(
                                    fieldSerializerSnapshot.restoreSerializer());
                        });

        final LinkedHashMap<Class<?>, TypeSerializer<?>> registeredSubclassSerializers =
                restoreSerializers(
                        snapshotData.getRegisteredSubclassSerializerSnapshots().unwrapOptionals());
        final Tuple2<LinkedHashMap<Class<?>, Integer>, TypeSerializer<Object>[]>
                decomposedSubclassSerializerRegistry =
                        decomposeSubclassSerializerRegistry(registeredSubclassSerializers);

        final LinkedHashMap<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializers =
                restoreSerializers(
                        snapshotData
                                .getNonRegisteredSubclassSerializerSnapshots()
                                .unwrapOptionals());

        return new PojoSerializer<>(
                snapshotData.getPojoClass(),
                restoredFields.toArray(new Field[numFields]),
                restoredFieldSerializers.toArray(new TypeSerializer[numFields]),
                decomposedSubclassSerializerRegistry.f0,
                decomposedSubclassSerializerRegistry.f1,
                nonRegisteredSubclassSerializers,
                new SerializerConfigImpl());
    }
@Test
    void testRestoreSerializerWithNewFields() {
        final PojoSerializerSnapshot<TestPojo> testSnapshot =
                buildTestSnapshot(Collections.singletonList(HEIGHT_FIELD));

        final TypeSerializer<TestPojo> restoredSerializer = testSnapshot.restoreSerializer();
        assertThat(restoredSerializer).isInstanceOf(PojoSerializer.class);
        final PojoSerializer<TestPojo> restoredPojoSerializer =
                (PojoSerializer<TestPojo>) restoredSerializer;

        final Field[] restoredFields = restoredPojoSerializer.getFields();
        assertThat(restoredFields).containsExactly(HEIGHT_FIELD.field);

        final TypeSerializer<?>[] restoredFieldSerializers =
                restoredPojoSerializer.getFieldSerializers();
        assertThat(restoredFieldSerializers).containsExactly(DoubleSerializer.INSTANCE);
    }