public static DataSchema avroToDataSchema(String avroSchemaInJson, AvroToDataSchemaTranslationOptions options)
    throws IllegalArgumentException
  {
    ValidationOptions validationOptions = SchemaParser.getDefaultSchemaParserValidationOptions();
    validationOptions.setAvroUnionMode(true);

    SchemaParserFactory parserFactory = SchemaParserFactory.instance(validationOptions);

    DataSchemaResolver resolver = getResolver(parserFactory, options);
    PegasusSchemaParser parser = parserFactory.create(resolver);
    parser.parse(avroSchemaInJson);
    if (parser.hasError())
    {
      throw new IllegalArgumentException(parser.errorMessage());
    }
    assert(parser.topLevelDataSchemas().size() == 1);
    DataSchema dataSchema = parser.topLevelDataSchemas().get(0);
    DataSchema resultDataSchema = null;

    AvroToDataSchemaTranslationMode translationMode = options.getTranslationMode();
    if (translationMode == AvroToDataSchemaTranslationMode.RETURN_EMBEDDED_SCHEMA ||
        translationMode == AvroToDataSchemaTranslationMode.VERIFY_EMBEDDED_SCHEMA)
    {
      // check for embedded schema

      Object dataProperty = dataSchema.getProperties().get(SchemaTranslator.DATA_PROPERTY);
      if (dataProperty != null && dataProperty.getClass() == DataMap.class)
      {
        Object schemaProperty = ((DataMap) dataProperty).get(SchemaTranslator.SCHEMA_PROPERTY);
        if (schemaProperty.getClass() == DataMap.class)
        {
          SchemaParser embeddedSchemaParser = SchemaParserFactory.instance().create(null);
          embeddedSchemaParser.parse(Arrays.asList(schemaProperty));
          if (embeddedSchemaParser.hasError())
          {
            throw new IllegalArgumentException("Embedded schema is invalid\n" + embeddedSchemaParser.errorMessage());
          }
          assert(embeddedSchemaParser.topLevelDataSchemas().size() == 1);
          resultDataSchema = embeddedSchemaParser.topLevelDataSchemas().get(0);

          if (translationMode == AvroToDataSchemaTranslationMode.VERIFY_EMBEDDED_SCHEMA)
          {
            // additional verification to make sure that embedded schema translates to Avro schema
            DataToAvroSchemaTranslationOptions dataToAvroSchemaOptions = new DataToAvroSchemaTranslationOptions();
            Object optionalDefaultModeProperty = ((DataMap) dataProperty).get(SchemaTranslator.OPTIONAL_DEFAULT_MODE_PROPERTY);
            dataToAvroSchemaOptions.setOptionalDefaultMode(OptionalDefaultMode.valueOf(optionalDefaultModeProperty.toString()));
            Schema avroSchemaFromEmbedded = dataToAvroSchema(resultDataSchema, dataToAvroSchemaOptions);
            Schema avroSchemaFromJson = AvroCompatibilityHelper.parse(avroSchemaInJson, SchemaParseConfiguration.STRICT, null).getMainSchema();
            Object embededSchemaPropertyVal = avroSchemaFromJson.getObjectProp(DATA_PROPERTY);
            if (embededSchemaPropertyVal != null)
            {
              avroSchemaFromEmbedded.addProp(DATA_PROPERTY, embededSchemaPropertyVal);
            }
            if (!avroSchemaFromEmbedded.equals(avroSchemaFromJson))
            {
              throw new IllegalArgumentException("Embedded schema does not translate to input Avro schema: " + avroSchemaInJson);
            }
          }
        }
      }
    }
    if (resultDataSchema == null)
    {
      // translationMode == TRANSLATE or no embedded schema

      DataSchemaTraverse traverse = new DataSchemaTraverse();
      traverse.traverse(dataSchema, AvroToDataSchemaConvertCallback.INSTANCE);
      // convert default values
      traverse.traverse(dataSchema, DefaultAvroToDataConvertCallback.INSTANCE);
      // make sure it can round-trip
      String dataSchemaJson = dataSchema.toString();
      resultDataSchema = DataTemplateUtil.parseSchema(dataSchemaJson);
    }
    return resultDataSchema;
  }
@Test
  public void testNullValueInSchemaParser() {
    String schemaWithNullValueForProperty =
        "{"
            + "\"type\":\"record\","
            + "\"name\":\"request\","
            + "\"namespace\":\"com.linkedin.test\","
            + "\"doc\":\"Doc\","
            + "\"fields\":["
                + "{"
                  + "\"name\":\"contentProviders\","
                  + "\"type\":["
                        + "\"null\","
                        + "{"
                          + "\"type\":\"array\","
                          + "\"items\":"
                            + "{"
                              + "\"type\":\"record\","
                              + "\"name\":\"ContentProviders\","
                              + "\"doc\":\"The Ids of the content provider\","
                          + "\"fields\":["
                          + "{\"name\":\"EntpProvider\","
                          + "\"type\":["
                          + "\"null\","
                          + "{"
                            + "\"type\":\"record\","
                            + "\"name\":\"EntpProvider\","
                            + "\"fields\":["
                            + "{"
                              + "\"name\":\"ID\","
                              + "\"type\":\"string\","
                              + "\"compliance\":{"
                              + "\"policy\":\"ENTERPRISE_ACCOUNT_ID\","
                              + "\"format\":\"ID\"}"
                            + "}"
                          + "]"
                        + "}"
                  + "],"
            + "\"doc\":\"The ID.\","
            + "\"default\":null,"
            + "\"compliance\":{"
              + "\"EntpProvider\":\"INHERITED\"}}],"
              + "\"compliance\":\"INHERITED\"},"
              + "\"default\":null,"
              + "\"compliance\":\"INHERITED\"}"
            + "],"
            + "\"doc\":\"Array of test Ids1\","
            + "\"default\":null,"
            + "\"compliance\":{\"array\":\"INHERITED\"}},"
            + "{"
                + "\"name\":\"enterpriseScopeFilterId\","
                + "\"type\":["
                + "\"null\","
                + "{"
                  + "\"type\":\"array\","
                  + "\"items\":\"string\"}],"
              + "\"doc\":\"A list of enterprise entities Ids\","
              + "\"default\":null,"
              + "\"compliance\":\"NONE\"}],"
              + "\"schemaType\":\"DocumentSchema\","
              + "\"version\":10,"
              + "\"upconvertVersion\":10,"
              + "\"evolutionSafetyMode\":\"IGNORE_WARNINGS\""
            + "}";
    Schema schema = Schema.parse(schemaWithNullValueForProperty);
    DataSchema dataSchema = SchemaTranslator.avroToDataSchema(schema);
    Assert.assertNotNull(dataSchema);
  }