private static <T> TypeSerializerSchemaCompatibility<T>
avroCompatibilityToFlinkCompatibility(SchemaPairCompatibility compatibility) {

	switch (compatibility.getType()) {
		case COMPATIBLE: {
			// The new serializer would be able to read data persisted with *this* serializer, therefore no migration
			// is required.
			return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
		}
		case INCOMPATIBLE: {
			return TypeSerializerSchemaCompatibility.incompatible();
		}
		case RECURSION_IN_PROGRESS:
		default:
			return TypeSerializerSchemaCompatibility.incompatible();
	}
}
 

private static <T> TypeSerializerSchemaCompatibility<T>
avroCompatibilityToFlinkCompatibility(SchemaPairCompatibility compatibility) {

	switch (compatibility.getType()) {
		case COMPATIBLE: {
			// The new serializer would be able to read data persisted with *this* serializer, therefore no migration
			// is required.
			return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
		}
		case INCOMPATIBLE: {
			return TypeSerializerSchemaCompatibility.incompatible();
		}
		case RECURSION_IN_PROGRESS:
		default:
			return TypeSerializerSchemaCompatibility.incompatible();
	}
}
 

private static <T> TypeSerializerSchemaCompatibility<T>
avroCompatibilityToFlinkCompatibility(SchemaPairCompatibility compatibility) {

	switch (compatibility.getType()) {
		case COMPATIBLE: {
			// The new serializer would be able to read data persisted with *this* serializer, therefore no migration
			// is required.
			return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
		}
		case INCOMPATIBLE: {
			return TypeSerializerSchemaCompatibility.incompatible();
		}
		case RECURSION_IN_PROGRESS:
		default:
			return TypeSerializerSchemaCompatibility.incompatible();
	}
}
 

/**
 * Resolves writer/reader schema compatibly.
 *
 * <p>Checks whenever a new version of a schema (reader) can read values serialized with the old schema (writer).
 * If the schemas are compatible according to {@code Avro} schema resolution rules
 * (@see <a href="https://avro.apache.org/docs/current/spec.html#Schema+Resolution">Schema Resolution</a>).
 */
@VisibleForTesting
static <T> TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
	Schema writerSchema,
	Schema readerSchema) {

	if (Objects.equals(writerSchema, readerSchema)) {
		return TypeSerializerSchemaCompatibility.compatibleAsIs();
	}

	final SchemaPairCompatibility compatibility =
		SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema);

	return avroCompatibilityToFlinkCompatibility(compatibility);
}
 

/**
 * Resolves writer/reader schema compatibly.
 *
 * <p>Checks whenever a new version of a schema (reader) can read values serialized with the old schema (writer).
 * If the schemas are compatible according to {@code Avro} schema resolution rules
 * (@see <a href="https://avro.apache.org/docs/current/spec.html#Schema+Resolution">Schema Resolution</a>).
 */
@VisibleForTesting
static <T> TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
	Schema writerSchema,
	Schema readerSchema) {

	if (Objects.equals(writerSchema, readerSchema)) {
		return TypeSerializerSchemaCompatibility.compatibleAsIs();
	}

	final SchemaPairCompatibility compatibility =
		SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema);

	return avroCompatibilityToFlinkCompatibility(compatibility);
}
 

/**
 * Resolves writer/reader schema compatibly.
 *
 * <p>Checks whenever a new version of a schema (reader) can read values serialized with the old schema (writer).
 * If the schemas are compatible according to {@code Avro} schema resolution rules
 * (@see <a href="https://avro.apache.org/docs/current/spec.html#Schema+Resolution">Schema Resolution</a>).
 */
@VisibleForTesting
static <T> TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
	Schema writerSchema,
	Schema readerSchema) {

	if (Objects.equals(writerSchema, readerSchema)) {
		return TypeSerializerSchemaCompatibility.compatibleAsIs();
	}

	final SchemaPairCompatibility compatibility =
		SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema);

	return avroCompatibilityToFlinkCompatibility(compatibility);
}