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 <agnelo.m.dco...@gmail.com>
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]*
>
