@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
    void nonValidSchemaEvaluationShouldResultInCompatibleSerializers() {
        final AvroSerializer<GenericRecord> originalSerializer =
                new AvroSerializer<>(GenericRecord.class, FIRST_REQUIRED_LAST_OPTIONAL);
        final AvroSerializer<GenericRecord> newSerializer =
                new AvroSerializer<>(GenericRecord.class, BOTH_REQUIRED);

        TypeSerializerSnapshot<GenericRecord> originalSnapshot =
                originalSerializer.snapshotConfiguration();

        assertThat(
                        newSerializer
                                .snapshotConfiguration()
                                .resolveSchemaCompatibility(originalSnapshot))
                .is(isIncompatible());
    }