[FLINK-25132] KafkaSource cannot work with object-reusing DeserializationSchema

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.14.0, 1.14.1, 1.15.0
• Fix Version/s: 1.14.3, 1.15.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Qingsheng Ren
• Reporter: Qingsheng Ren
• Created: 01/Dec/21 09:57
• Updated: 09/Jun/23 16:47
• Resolved: 27/Dec/21 02:15

Description

Currently Kafka source deserializes ConsumerRecords in split reader and puts them into the elementQueue, then task's main thread polls these records from the queue asynchronously. This mechanism cannot cooperate with DeserializationSchemas with object reuse: all records staying in the element queue points to the same object.

A solution would be moving deserialization to RecordEmitter, which works in the task's main thread. 

Notes that this issue actually effects all sources which do deserialization in split reader. 

Issue Links

• causes: FLINK-32303 Incorrect error message in KafkaSource
• relates to: FLINK-28083 PulsarSource cannot work with object-reusing DeserializationSchema.
• links to: GitHub Pull Request #17991