@Override
    public TypeSerializer<T> restoreSerializer() {
        checkNotNull(runtimeType);
        checkNotNull(schema);

        if (runtimeSchema != null) {
            return new AvroSerializer<>(
                    runtimeType,
                    new SerializableAvroSchema(runtimeSchema),
                    new SerializableAvroSchema(schema));
        } else {
            return new AvroSerializer<>(
                    runtimeType,
                    new SerializableAvroSchema(schema),
                    new SerializableAvroSchema(schema));
        }
    }
@Test
    void recordSerializedShouldBeDeserializeWithTheResortedSerializer() throws IOException {
        // user is an avro generated test object.
        final User user = TestDataGenerator.generateRandomUser(new Random());
        final AvroSerializer<User> originalSerializer = new AvroSerializer<>(User.class);
        //
        // first serialize the record
        //
        ByteBuffer serializedUser = serialize(originalSerializer, user);
        //
        // then restore a serializer from the snapshot
        //
        TypeSerializer<User> restoredSerializer =
                originalSerializer.snapshotConfiguration().restoreSerializer();
        //
        // now deserialize the user with the resorted serializer.
        //
        User restoredUser = deserialize(restoredSerializer, serializedUser);

        assertThat(restoredUser).isEqualTo(user);
    }