@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 aLargeSchemaAvroSnapshotIsCompatibleAfterARoundTrip() throws IOException {
        // construct the large schema up to a size of 65535 bytes.
        int thresholdSize = 65535;
        StringBuilder schemaField = new StringBuilder(thresholdSize);
        for (int i = 0; i <= thresholdSize; i++) {
            schemaField.append('a');
        }
        Schema largeSchema =
                SchemaBuilder.record("name")
                        .namespace("org.apache.flink")
                        .fields()
                        .requiredString(schemaField.toString())
                        .endRecord();

        AvroSerializer<GenericRecord> serializer =
                new AvroSerializer<>(GenericRecord.class, largeSchema);
        AvroSerializerSnapshot<GenericRecord> restored =
                roundTrip(serializer.snapshotConfiguration());

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