AvroOverride createFromDataSchema(DataSchema schema)
  {
    AvroOverride avroOverride = null;
    Map<String, Object> properties = schema.getProperties();
    Object avro = properties.get("avro");
    if (avro != null)
    {
      boolean ok = true;
      Object avroTranslator = null;
      Object avroSchema = null;
      if (avro.getClass() != DataMap.class)
      {
        emitMessage("%1$s has \"avro\" property whose value is not a JSON object", schema);
        ok = false;
      }
      else
      {
        DataMap avroMap = (DataMap) avro;
        avroTranslator = avroMap.get("translator");
        avroSchema = avroMap.get("schema");
      }

      // process "schema" property

      String avroSchemaFullName = null;
      DataMap avroSchemaDataMap = null;
      if (avroSchema != null)
      {
        if (avroSchema.getClass() != DataMap.class)
        {
          emitMessage("\"schema\" property is not a JSON object, value is %1$s", avroSchema);
          ok = false;
        }
        else
        {
          avroSchemaDataMap = (DataMap) avroSchema;
          Object name = avroSchemaDataMap.get("name");
          if (name == null)
          {
            emitMessage("\"name\" property of \"schema\" property is required");
            ok = false;
          }
          else if (name.getClass() != String.class)
          {
            emitMessage("\"name\" property of \"schema\" property is not a string, value is %1$s", name);
            ok = false;
          }
          else
          {
            Object namespace = avroSchemaDataMap.get("namespace");
            if (namespace != null)
            {
              if (namespace.getClass() != String.class)
              {
                emitMessage("\"namespace\" property of \"schema\" property is not a string, value is %1$s", namespace);
                ok = false;
              }
              else if (Name.isValidNamespace((String) namespace) == false)
              {
                emitMessage("%1$s is not a valid namespace", namespace);
                ok = false;
              }
            }
            avroSchemaFullName = namespace == null ? (String) name : namespace + "." + name;
          }
        }
      }

      // process "translator" property

      String customDataTranslatorClassName = null;
      CustomDataTranslator customDataTranslator = null;
      if (avroTranslator != null)
      {
        if (avroTranslator.getClass() != DataMap.class)
        {
          emitMessage("\"translator\" property is not a JSON object, value is %1$s", avroTranslator);
          ok = false;
        }
        else
        {
          DataMap avroTranslatorMap = (DataMap) avroTranslator;
          Object avroTranslatorClass = avroTranslatorMap.get("class");
          if (avroTranslatorClass == null)
          {
            emitMessage("\"class\" property of \"translator\" property is required");
            ok = false;
          }
          else if (avroTranslatorClass.getClass() != String.class)
          {
            emitMessage("\"class\" property of \"translator\" property is not a string, value is %1$s", avroTranslatorClass);
            ok = false;
          }
          else
          {
            customDataTranslatorClassName = (String) avroTranslatorClass;
            if (_instantiateCustomDataTranslator)
            {
              try
              {
                Class<?> translatorClass = Class.forName(customDataTranslatorClassName, true, Thread.currentThread().getContextClassLoader());
                customDataTranslator = (CustomDataTranslator) translatorClass.getDeclaredConstructor().newInstance();
              }
              catch (ClassCastException e)
              {
                emitMessage("%1$s is not a %2$s", customDataTranslatorClassName, CustomDataTranslator.class.getName());
                ok = false;
              }
              catch (ClassNotFoundException e)
              {
                emitMessage("%1$s class not found", customDataTranslatorClassName);
                ok = false;
              }
              catch (IllegalAccessException | InstantiationException | InvocationTargetException | NoSuchMethodException e)
              {
                emitMessage("%1$s cannot be instantiated due to %2$s", customDataTranslatorClassName, e.getClass().getName());
                ok = false;
              }
            }
          }
        }
      }

      if (ok)
      {
        // make both "schema" and "translator" are present if either is present
        if ((avroSchemaFullName != null && customDataTranslatorClassName == null) ||
            (avroSchemaFullName == null && customDataTranslatorClassName != null))
        {
          emitMessage("both \"translator\" and \"schema\" properties of \"avro\" are required if either is present");
          ok = false;
        }
        if (schema.getType() != DataSchema.Type.RECORD)
        {
          emitMessage("%1$s has \"avro\" properties \"translator\" and \"schema\" that are only applicable to records", schema);
          ok = false;
        }
      }
      if (ok)
      {
        avroOverride = new AvroOverride(avroSchemaFullName, avroSchemaDataMap, customDataTranslatorClassName, customDataTranslator);
      }
    }
    return avroOverride;
  }
@Test
  public void testGood() throws IOException
  {
    String schemaText =
      "{\n" +
      "  \"type\" : \"record\",\n" +
      "  \"name\" : \"Foo\",\n" +
      "  \"fields\" : [],\n" +
      "  \"avro\" : {\n" +
      "    \"schema\" : {\n" +
      "      \"name\" : \"CustomFoo\",\n" +
      "      \"namespace\" : \"org.foobar\"\n" +
      "    },\n" +
      "    \"translator\" : {\n" +
      "      \"class\" : \"" + MyCustomDataTranslator.class.getName() + "\"\n" +
      "    }\n" +
      "  }\n" +
      "}";

    final boolean debug = false;

    MyAvroOverrideFactory avroOverrideFactory = new MyAvroOverrideFactory();

    DataSchema schema = TestUtil.dataSchemaFromString(schemaText);
    AvroOverride avroOverride = avroOverrideFactory.createFromDataSchema(schema);
    String message = avroOverrideFactory._messageList.toString();

    assertTrue(message.isEmpty());
    assertSame(avroOverride.getAvroSchemaDataMap(), ((DataMap) schema.getProperties().get("avro")).get("schema"));
    assertEquals(avroOverride.getAvroSchemaFullName(), "org.foobar.CustomFoo");

    assertSame(avroOverride.getCustomDataTranslatorClassName(), ((DataMap) ((DataMap) schema.getProperties().get("avro")).get("translator")).get("class"));
    assertSame(avroOverride.getCustomDataTranslator().getClass(), MyCustomDataTranslator.class);
  }