public static SchemaPairCompatibility checkReaderWriterCompatibility(final Schema reader, final Schema writer) {
    final SchemaCompatibilityResult compatibility = new ReaderWriterCompatibilityChecker().getCompatibility(reader,
        writer);

    final String message;
    switch (compatibility.getCompatibility()) {
    case INCOMPATIBLE: {
      message = String.format(
          "Data encoded using writer schema:%n%s%n" + "will or may fail to decode using reader schema:%n%s%n",
          writer.toString(true), reader.toString(true));
      break;
    }
    case COMPATIBLE: {
      message = READER_WRITER_COMPATIBLE_MESSAGE;
      break;
    }
    default:
      throw new AvroRuntimeException("Unknown compatibility: " + compatibility);
    }

    return new SchemaPairCompatibility(compatibility, reader, writer, message);
  }
@Test
  void unionWriterSimpleReaderIncompatibility() {
    Schema mandatorySchema = SchemaBuilder.record("Account").fields().name("age").type().intType().noDefault()
        .endRecord();
    Schema optionalSchema = SchemaBuilder.record("Account").fields().optionalInt("age").endRecord();

    SchemaPairCompatibility compatibility = checkReaderWriterCompatibility(mandatorySchema, optionalSchema);

    assertEquals(SchemaCompatibilityType.INCOMPATIBLE, compatibility.getType());

    Incompatibility incompatibility = compatibility.getResult().getIncompatibilities().get(0);
    assertEquals("reader type: INT not compatible with writer type: NULL", incompatibility.getMessage());
    assertEquals("/fields/0/type/0", incompatibility.getLocation());
  }