@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 restorePastSnapshots() throws IOException {
        for (int pastVersion : PAST_VERSIONS) {
            AvroSerializer<GenericRecord> currentSerializer =
                    new AvroSerializer<>(GenericRecord.class, Address.getClassSchema());

            DataInputView in =
                    new DataInputDeserializer(
                            Files.readAllBytes(getSerializerSnapshotFilePath(pastVersion)));

            TypeSerializerSnapshot<GenericRecord> restored =
                    TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot(
                            in, AvroSerializer.class.getClassLoader());

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