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 testAvroPartialDefaultFields() throws IOException
  {
    String schemaWithPartialDefaultFields = "{" +
        "  \"type\": \"record\"," +
        "  \"name\": \"testRecord\"," +
        "  \"fields\": [" +
        "    {" +
        "      \"name\": \"recordFieldWithDefault\"," +
        "      \"type\": {" +
        "        \"type\": \"record\"," +
        "        \"name\": \"recordType\"," +
        "        \"fields\": [" +
        "          {" +
        "            \"name\": \"mapField\"," +
        "            \"type\": {" +
        "              \"type\": \"map\"," +
        "              \"values\": \"string\"" +
        "            }" +
        "          }," +
        "          {" +
        "            \"name\": \"optionalRecordField\"," +
        "            \"type\": [" +
        "              \"null\"," +
        "              {" +
        "                \"type\": \"record\"," +
        "                \"name\": \"simpleRecordType\"," +
        "                \"fields\": [" +
        "                  {" +
        "                    \"name\": \"stringField\"," +
        "                    \"type\": \"string\"" +
        "                  }" + "                ]" +
        "              }" +
        "            ]," +
        "            \"default\": null" +
        "          }" +
        "        ]" +
        "      }," +
        "      \"default\": {" +
        "        \"mapField\": {}" +
        "      }" +
        "    }" +
        "  ]" +
        "}";

    Schema schema = Schema.parse(schemaWithPartialDefaultFields);
    DataSchema dataSchema = SchemaTranslator.avroToDataSchema(schema);
    Assert.assertNotNull(dataSchema);
  }