Re: In Flink SQL for kafka avro based table , is there support for FORWARD_TRANSITIVE schema change?

Posted by Agnelo Dcosta on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/In-Flink-SQL-for-kafka-avro-based-table-is-there-support-for-FORWARD-TRANSITIVE-schema-change-tp42939p42941.html

Hi Arvid, 

> writer schema encoded if you are using no schema registry?
on the producer side we are using node with https://registry.npmjs.org/avsc/-/avsc-5.5.3.tgz

and https://registry.npmjs.org/sinek/-/sinek-9.1.0.tgz libraries to publish messages. We specify the avro schema file to encode messages in avro format.

>Flink know with which schema the data has been written so that it can map it to the new schema?

With 1.11 we used to specify the schema file as part of the flink sql table definition.

However with 1.12 the schema is derived from the message/table definition. We do not specify any schema as such.

On Tue, Apr 13, 2021 at 11:58 PM Arvid Heise <[hidden email]> wrote:

Hi Agnelo,

How is the writer schema encoded if you are using no schema registry? Or phrased differently: how does Flink know with which schema the data has been written so that it can map it to the new schema?

On Wed, Apr 14, 2021 at 8:33 AM Agnelo Dcosta <[hidden email]> wrote:

Hi, we are using Flink SQL 1.12 and have a couple of tables created from kafka topics. Format is avro (not confluent avro) and no schema registry as such.

In flink 1.11 we used to specify the schema, however in 1.12 the schema is derived from the message itself.

Is it possible for the producers to start sending new fields without changes in the flink app?

 

For example :

{

  "name": "topic1",

  "type": "record",

  "fields": [

      {

      "name": "field1",

      "type": "string"

    },

{

      "name": "field2",

      "type": "string"

    },

{

      "name": "field3",

      "type": "string"

    },

]

}

 

Flink table has:

CREATE TABLE topic1(\n"

                + " field1 string not null \n"

                + " ,field2 string not null \n"    

"'connector' = 'kafka' \n"

             + ",'topic' = 'topic1' \n"

             + ",'scan.startup.mode' = 'latest-offset' \n"

             + ",'properties.group.id' = 'topic1' \n"

             + ",'properties.bootstrap.servers' = 'localhost:8082' \n"

              + ",'properties.enable.auto.commit' = 'true' \n"

             + ",'format' = 'avro' \n";

 

With above settings I get a deserialization error:

 

java.io.IOException: Failed to deserialize Avro record.

        at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:104) ~[flink-sql-avro-1.12.0.jar:1.12.0]

        at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:44) ~[flink-sql-avro-1.12.0.jar:1.12.0]

        at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82) ~[flink-core-1.12.0.jar:1.12.0]

        at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.deserialize(DynamicKafkaDeserializationSchema.java:113) ~[flink-connector-kafka_2.11-1.12.0.jar:1.12.0]

        at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:177) ~[flink-connector-kafka_2.11-1.12.0.jar:1.12.0]

        at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.runFetchLoop(KafkaFetcher.java:137) ~[flink-connector-kafka_2.11-1.12.0.jar:1.12.0]

        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:761) ~[flink-connector-kafka_2.11-1.12.0.jar:1.12.0]

        at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) ~[flink-streaming-java_2.11-1.12.0.jar:1.12.0]

        at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) ~[flink-streaming-java_2.11-1.12.0.jar:1.12.0]