import org.apache.avro.SchemaCompatibility; //导入依赖的package包/类
/**
 * It is valid when using a canBeReadStrategy for messages to arrive with fields not declared in the earlier
 * schema.
 */
@Test
public void verifyAvroBehaviour_SchemaCompatibility() throws SchemaValidationException {
  Schema newSchema = SchemaBuilder.record("a").fields().requiredString("x").requiredString("y").endRecord();
  Schema oldSchema = SchemaBuilder.record("a").fields().requiredString("x").endRecord();
  SchemaCompatibilityType compatibilityType = SchemaCompatibility
      .checkReaderWriterCompatibility(oldSchema, newSchema)
      .getType();
  assertThat(compatibilityType, is(SchemaCompatibilityType.COMPATIBLE));
}
 

import org.apache.avro.SchemaCompatibility; //导入依赖的package包/类
@Override
@SuppressWarnings("deprecation")
public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot) {
	if (configSnapshot instanceof AvroSchemaSerializerConfigSnapshot) {
		// proper schema snapshot, can do the sophisticated schema-based compatibility check
		final String schemaString = ((AvroSchemaSerializerConfigSnapshot) configSnapshot).getSchemaString();
		final Schema lastSchema = new Schema.Parser().parse(schemaString);

		checkAvroInitialized();
		final SchemaPairCompatibility compatibility =
				SchemaCompatibility.checkReaderWriterCompatibility(schema, lastSchema);

		return compatibility.getType() == SchemaCompatibilityType.COMPATIBLE ?
				CompatibilityResult.compatible() : CompatibilityResult.requiresMigration();
	}
	else if (configSnapshot instanceof AvroSerializerConfigSnapshot) {
		// old snapshot case, just compare the type
		// we don't need to restore any Kryo stuff, since Kryo was never used for persistence,
		// only for object-to-object copies.
		final AvroSerializerConfigSnapshot old = (AvroSerializerConfigSnapshot) configSnapshot;
		return type.equals(old.getTypeClass()) ?
				CompatibilityResult.compatible() : CompatibilityResult.requiresMigration();
	}
	else {
		return CompatibilityResult.requiresMigration();
	}
}
 

import org.apache.avro.SchemaCompatibility; //导入依赖的package包/类
public SchemaCompatiblityCheck schemaBackwardCompatibility(SchemaDTO schemaDto) {

    String schemaContent = schemaDto.getContents();

    SchemaCompatibility.SchemaPairCompatibility schemaCompatibility;
    Schema writer;

    TypedQuery<SchemaTopics> query = em.createNamedQuery(
        "SchemaTopics.findByName", SchemaTopics.class);
    query.setParameter("name", schemaDto.getName());

    try {
      Schema reader = new Schema.Parser().parse(schemaContent);

      for (SchemaTopics schemaTopic : query.getResultList()) {

        writer = new Schema.Parser().parse(schemaTopic.getContents());

        schemaCompatibility = SchemaCompatibility.
            checkReaderWriterCompatibility(reader, writer);

        switch (schemaCompatibility.getType()) {

          case COMPATIBLE:
            break;
          case INCOMPATIBLE:
            return SchemaCompatiblityCheck.INCOMPATIBLE;
          case RECURSION_IN_PROGRESS:
            break;
        }
      }
    } catch (SchemaParseException ex) {
      return SchemaCompatiblityCheck.INVALID;
    }
    return SchemaCompatiblityCheck.COMPATIBLE;
  }
 

import org.apache.avro.SchemaCompatibility; //导入依赖的package包/类
/**
 * keySchema is valid if a record with newestSchema can be converted to a record with keySchema.
 */
private boolean isKeySchemaValid(Schema keySchema, Schema topicSchema) {
  return SchemaCompatibility.checkReaderWriterCompatibility(keySchema, topicSchema).getType()
      .equals(SchemaCompatibilityType.COMPATIBLE);
}
 

import org.apache.avro.SchemaCompatibility; //导入依赖的package包/类
/**
 * keySchema is valid if a record with newestSchema can be converted to a record with keySchema.
 */
public static boolean isKeySchemaValid(Schema keySchema, Schema topicSchema) {
  return SchemaCompatibility.checkReaderWriterCompatibility(keySchema, topicSchema).getType()
      .equals(SchemaCompatibilityType.COMPATIBLE);
}