@Override
    public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
            TypeSerializerSnapshot<T> oldSerializerSnapshot) {
        if (!(oldSerializerSnapshot instanceof AvroSerializerSnapshot)) {
            return TypeSerializerSchemaCompatibility.incompatible();
        }
        AvroSerializerSnapshot<?> oldAvroSerializerSnapshot =
                (AvroSerializerSnapshot<?>) oldSerializerSnapshot;
        return resolveSchemaCompatibility(oldAvroSerializerSnapshot.schema, schema);
    }
@Test
    @SuppressWarnings({"unchecked", "rawtypes"})
    public void
            changingFromGenericToSpecificWithCompatibleSchemaShouldResultInCompatibleSerializers() {
        // starting with a generic serializer
        AvroSerializer<Object> generic = new AvroSerializer(GenericRecord.class, User.SCHEMA$);
        TypeSerializerSnapshot<Object> genericSnapshot = generic.snapshotConfiguration();

        // then upgrading to a specific serializer
        AvroSerializer<Object> specificSerializer = new AvroSerializer(User.class);
        specificSerializer.snapshotConfiguration();

        assertThat(
                        specificSerializer
                                .snapshotConfiguration()
                                .resolveSchemaCompatibility(genericSnapshot))
                .is(isCompatibleAsIs());
    }