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 testBad() throws IOException
  {
    final String[][] inputs =
      {
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : 0\n" +
          "}",
          "has \"avro\" property whose value is not a JSON object"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"schema\" : 0\n" +
          "  }\n" +
          "}",
          "\"schema\" property is not a JSON object, value is 0"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"schema\" : {\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "\"name\" property of \"schema\" property is required"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"schema\" : {\n" +
          "      \"name\" : 0\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "\"name\" property of \"schema\" property is not a string, value is 0"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"schema\" : {\n" +
          "      \"name\" : \"CustomFoo\",\n" +
          "      \"namespace\" : 0\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "\"namespace\" property of \"schema\" property is not a string, value is 0"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"translator\" : 0\n" +
          "  }\n" +
          "}",
          "\"translator\" property is not a JSON object, value is 0"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"translator\" : {\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "\"class\" property of \"translator\" property is required"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"translator\" : {\n" +
          "      \"class\" : 0\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "\"class\" property of \"translator\" property is not a string, value is 0"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"schema\" : {\n" +
          "      \"name\" : \"CustomFoo\",\n" +
          "      \"namespace\" : \"org.foobar\"\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "both \"translator\" and \"schema\" properties of \"avro\" are required if either is present"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"translator\" : {\n" +
          "      \"class\" : \"" + MyCustomDataTranslator.class.getName() + "\"\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "both \"translator\" and \"schema\" properties of \"avro\" are required if either is present"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"schema\" : {\n" +
          "      \"name\" : \"CustomFoo\",\n" +
          "      \"namespace\" : \"org.foobar\"\n" +
          "    },\n" +
          "    \"translator\" : {\n" +
          "      \"class\" : \"java.util.Date\"\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "java.util.Date is not a com.linkedin.data.avro.CustomDataTranslator"
        },
        {
          "{\n" +
          "  \"type\" : \"record\",\n" +
          "  \"name\" : \"Foo\",\n" +
          "  \"fields\" : [],\n" +
          "  \"avro\" : {\n" +
          "    \"schema\" : {\n" +
          "      \"name\" : \"CustomFoo\",\n" +
          "      \"namespace\" : \"org.foobar\"\n" +
          "    },\n" +
          "    \"translator\" : {\n" +
          "      \"class\" : \"" + PrivateConstructorMyCustomDataTranslator.class.getName() + "\"\n" +
          "    }\n" +
          "  }\n" +
          "}",
          "com.linkedin.data.avro.TestAvroOverrideFactory$PrivateConstructorMyCustomDataTranslator cannot be instantiated due to java.lang.IllegalAccessException"
        },
      };

    final boolean debug = false;

    MyAvroOverrideFactory avroOverrideFactory = new MyAvroOverrideFactory();


    for (String[] row : inputs)
    {
      String schemaText = row[0];

      DataSchema schema = TestUtil.dataSchemaFromString(schemaText);
      AvroOverride avroOverride = avroOverrideFactory.createFromDataSchema(schema);

      String message = avroOverrideFactory._messageList.toString();
      if (debug) TestUtil.out.println(message);

      for (int i = 1; i < row.length; i++)
      {
        assertTrue(message.contains(row[i]), message + " contains " + row[i]);
      }
      avroOverrideFactory.reset();
    }
  }