@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 anAvroSnapshotIsCompatibleAfterARoundTrip() throws IOException {
        AvroSerializer<GenericRecord> serializer =
                new AvroSerializer<>(GenericRecord.class, FIRST_REQUIRED_LAST_OPTIONAL);

        AvroSerializerSnapshot<GenericRecord> restored =
                roundTrip(serializer.snapshotConfiguration());

        assertThat(serializer.snapshotConfiguration().resolveSchemaCompatibility(restored))
                .is(isCompatibleAsIs());
    }