title: AvroTypeException: Not an enum: MOBILE on DataFileWriter
body: <p>I am getting the following error message when I tried to write avro records using build-in <code>AvroKeyValueSinkWriter</code> in Flink 1.3.2 and avro 1.8.2:</p>
  
  <p>My schema looks like this:</p>
  
  <pre><code>{"namespace": "com.base.avro",
   "type": "record",
   "name": "Customer",
   "doc": "v6",
   "fields": [
       {"name": "CustomerID", "type": "string"},
       {"name": "platformAgent", "type": {
         "type": "enum",
         "name": "PlatformAgent",
         "symbols": ["WEB", "MOBILE", "UNKNOWN"]
         }, "default":"UNKNOWN"}
   ]
  }
  </code></pre>
  
  <p>And I am calling the following Flink code to write data:</p>
  
  <pre><code>    var properties = new util.HashMap[String, String]()
      val stringSchema = Schema.create(Type.STRING)
      val myTypeSchema = Customer.getClassSchema
      val keySchema = stringSchema.toString
      val valueSchema = myTypeSchema.toString
  
      val compress = true
      properties.put(AvroKeyValueSinkWriter.CONF_OUTPUT_KEY_SCHEMA, keySchema)
      properties.put(AvroKeyValueSinkWriter.CONF_OUTPUT_VALUE_SCHEMA, valueSchema)
      properties.put(AvroKeyValueSinkWriter.CONF_COMPRESS, compress.toString)
      properties.put(AvroKeyValueSinkWriter.CONF_COMPRESS_CODEC, DataFileConstants.SNAPPY_CODEC)
  
      val sink = new BucketingSink[org.apache.flink.api.java.tuple.Tuple2[String, Customer]]("s3://test/flink")
      sink.setBucketer(new DateTimeBucketer("yyyy-MM-dd/HH/mm/"))
      sink.setInactiveBucketThreshold(120000) // this is 2 minutes
      sink.setBatchSize(1024 * 1024 * 64) // this is 64 MB,
      sink.setPendingSuffix(".avro")
  
      val writer = new AvroKeyValueSinkWriter[String, Customer](properties)
      sink.setWriter(writer.duplicate())
  </code></pre>
  
  <p>However, it throws the following errors:</p>
  
  <pre><code>Caused by: org.apache.avro.AvroTypeException: Not an enum: MOBILE
      at org.apache.avro.generic.GenericDatumWriter.writeEnum(GenericDatumWriter.java:177)
      at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:119)
      at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
      at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:166)
      at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:156)
      at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:118)
      at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
      at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:166)
      at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:156)
      at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:118)
      at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
      at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:62)
      at org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:302)
      ... 10 more
  </code></pre>
  
  <p>Please suggest!</p>
  
  <p><strong>UPDATE 1:</strong> 
  I found this is kind of bug in avro 1.8+ based on this ticket: <a href="https://issues-test.apache.org/jira/browse/AVRO-1810" rel="nofollow noreferrer">https://issues-test.apache.org/jira/browse/AVRO-1810</a></p>