[FLINK-19629] Fix NullPointException when deserializing map field with null value for Avro format

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.11.2
• Fix Version/s: 1.11.3, 1.12.0
• Component/s: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
• Labels: pull-request-available
• Assignee: shizhengchao
• Reporter: shizhengchao
• Created: 14/Oct/20 03:18
• Updated: 22/Oct/20 12:31
• Resolved: 22/Oct/20 12:31

Description

create table tableA (
name STRING,
hobly MAP<STRING, STRING>,
phone STRING
) with (
'connector' = 'kafka-0.11',
'topic' = 'ShizcTest',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'ShizcTest',
'scan.startup.mode' = 'earliest-offset',
'format' = 'avro'
);
if hobly have an null value like this:

{"name": "shizc", "hobly":

{"key1":null}

, "phone": "1104564"}

cause an NullPointException:

java.io.IOException: Failed to deserialize Avro record.
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:150)
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:75)
	at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:81)
	at org.apache.flink.streaming.connectors.kafka.internals.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:56)
	at org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher.runFetchLoop(Kafka010Fetcher.java:147)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201)
Caused by: java.lang.NullPointerException: null
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.lambda$createConverter$57e941b$5(AvroRowDataDeserializationSchema.java:252)
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.lambda$createMapConverter$7941d275$1(AvroRowDataDeserializationSchema.java:315)
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.lambda$createNullableConverter$c3bac5d8$1(AvroRowDataDeserializationSchema.java:221)
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.lambda$createRowConverter$80d8b6bd$1(AvroRowDataDeserializationSchema.java:206)
	at org.apache.flink.formats.avro.AvroRowDataDeserializationSchema.deserialize(AvroRowDataDeserializationSchema.java:148)
	... 8 common frames omitted

Issue Links

• duplicates: FLINK-19622 Flinksql version 1.11 is for the NullPointerException of the Map type value value in the avro format
• links to: GitHub Pull Request #13634
• links to: GitHub Pull Request #13743