public static void checkSchemaCompatible(
      Schema tableSchema,
      Schema writerSchema,
      boolean shouldValidate,
      boolean allowProjection,
      Set<String> dropPartitionColNames) throws SchemaCompatibilityException {

    if (!allowProjection) {
      List<Schema.Field> missingFields = findMissingFields(tableSchema, writerSchema, dropPartitionColNames);
      if (!missingFields.isEmpty()) {
        throw new MissingSchemaFieldException(missingFields.stream().map(Schema.Field::name).collect(Collectors.toList()), writerSchema, tableSchema);
      }
    }

    // TODO(HUDI-4772) re-enable validations in case partition columns
    //                 being dropped from the data-file after fixing the write schema
    if (dropPartitionColNames.isEmpty() && shouldValidate) {
      AvroSchemaCompatibility.SchemaPairCompatibility result =
          AvroSchemaCompatibility.checkReaderWriterCompatibility(writerSchema, tableSchema, true);
      if (result.getType() != AvroSchemaCompatibility.SchemaCompatibilityType.COMPATIBLE) {
        throw new SchemaBackwardsCompatibilityException(result, writerSchema, tableSchema);
      }
    }
  }
@Test
  public void  testBrokenSchema() {
    assertThrows(SchemaBackwardsCompatibilityException.class,
        () -> AvroSchemaUtils.checkSchemaCompatible(FULL_SCHEMA, BROKEN_SCHEMA, true, false, Collections.emptySet()));
  }