[jira] [Created] (FLINK-28467) AVRO Table API fails during schema evolution

["Jaya Ananthram (Jira)" <ji...@apache.org>]

Jaya Ananthram created FLINK-28467:
--------------------------------------

             Summary: AVRO Table API fails during schema evolution
                 Key: FLINK-28467
                 URL: https://issues.apache.org/jira/browse/FLINK-28467
             Project: Flink
          Issue Type: Bug
          Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
    Affects Versions: 1.14.4
            Reporter: Jaya Ananthram
         Attachments: image-2022-07-08-15-01-49-648.png

It looks like the Flink Table API fails during the schema evolution. ie - when we add a new optional field in the producer side, the consumer (Flink table API) fails to parse the message with the old schema. Following are the exact scenario to reproduce,
 # Create a schema X with two field
 # Send five messages to Kafka using schema X
 # Update the schema X to add one new optional field with default NULL (at last position)
 # Send five messages to Kafka using schema Y
 # Create a Flink SQL job to consume all the 10 messages using schema X (with two fields)
 # Exactly it will fail at the 7th message to get the exception *_Malformed data. Length is negative: -xx_* (the 6th message will pass successfully though).

The complete stack trace is available below,
{code:java}
Caused by: org.apache.avro.AvroRuntimeException: Malformed data. Length is negative: -56	at org.apache.avro.io.BinaryDecoder.readString(BinaryDecoder.java:285)	at org.apache.avro.io.ResolvingDecoder.readString(ResolvingDecoder.java:208)	at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:469)	at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:459)	at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:191)	at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160)	at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:259)	at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:247)	at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179)	at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160)	at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)	at org.apache.flink.formats.avro.AvroDeserializationSchema.deserialize(AvroDeserializationSchema.java:142)	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:103)	... 19 more {code}
 

From [Avro specification|https://avro.apache.org/docs/1.10.0/spec.html] (refer to the section "Single object Encoding" or the attached image), it looks by default Avro provides the schema evolution support, so, as an end-user I expect the Flink table API to provide the same functionalities. I can also confirm that the old schema is able to decode all the 10 messages outside of Flink (ie - using simple hello world AvroDeserializer)

I am adding the root cause as a comment, as I am exactly sure whether my finding is correct.

*Note:* I am marking this as a Major ticket as we have another open ticket ([here|https://issues.apache.org/jira/browse/FLINK-20091]) which lacks the functionalities to  ignore failures. This means that, even if the user is ready to miss some message (when a batch contains two types of messages), still we can't specify the property to ignore it (to update the DDL once it crosses the switch over messages). So it looks like the Avro table API can't be used in PROD when we expect to change the schema. I assume most of the cases, we are expected to change schema. If the severity is not correct please feel free to reduce the priority.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)