[jira] [Comment Edited] (FLINK-33934) Flink SQL Source use raw format maybe lead to data lost

["Cai Liuyang (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-33934?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17800273#comment-17800273 ] 

Cai Liuyang edited comment on FLINK-33934 at 12/25/23 11:41 AM:
----------------------------------------------------------------

[~hackergin] Yeah,  turn off Object Reuse for all DeserializeSchema(that will be used by SourceOperator) is the simple way to avoid this data lost problem.

------ update ------

after review kafka source code, kafka will deserialize from ConsumerRecord to real object in SourceOperator MainThread, So it will not encounter this problem, see code: [KafkaRecordEmitter code|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaRecordEmitter.java#L53C43-L53C43] 

 

So, this problem will only appear in which case that DeserializeSchema::deserialize is called in FetcherThread

 


was (Author: cailiuyang):
[~hackergin] Yeah,  turn off Object Reuse for all DeserializeSchema(that will be used by SourceOperator) is the simple way to avoid this data lost problem.

> Flink SQL Source use raw format maybe lead to data lost
> -------------------------------------------------------
>
>                 Key: FLINK-33934
>                 URL: https://issues.apache.org/jira/browse/FLINK-33934
>             Project: Flink
>          Issue Type: Bug
>          Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table SQL / Runtime
>            Reporter: Cai Liuyang
>            Priority: Major
>
> In our product we encounter a case that lead to data lost, the job info: 
>    1. using flinkSQL that read data from messageQueue and write to hive (only select value field, doesn't contain metadata field)
>    2. the format of source table is raw format
>  
> But if we select value field and metadata field at the same time, than the data lost will not appear
>  
> After we review the code, we found that the reason is the object reuse of Raw-format(see code [RawFormatDeserializationSchema|https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java#L62]), why object reuse will lead to this problem is below (take kafka as example):
>     1. RawFormatDeserializationSchema will be used in the Fetcher-Thread of SourceOperator, Fetcher-Thread will read and deserialize data from kafka partition, than put data to ElementQueue (see code [SourceOperator FetcherTask |https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/FetchTask.java#L64])
>     2. SourceOperator's main thread will pull data from the ElementQueue(which is shared with the FetcherThread) and process it (see code [SourceOperator main thread|https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java#L188])
>     3. For RawFormatDeserializationSchema, its deserialize function will return the same object([reuse rowData object|https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java#L62])
>     4. So, if elementQueue have element that not be consumed, than the fetcherThread can change the filed of the reused rawData that RawFormatDeserializationSchema::deserialize returned, this will lead to data lost;
>  
> The reason that we select value and metadata field at the same time will not encounter data lost is:
>    if we select metadata field there will return a new RowData object see code: [DynamicKafkaDeserializationSchema deserialize with metadata field |https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaDeserializationSchema.java#L249] and if we only select value filed, it will reuse the RowData object that formatDeserializationSchema returned see code [DynamicKafkaDeserializationSchema deserialize only with value field|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaDeserializationSchema.java#L113]
>  
> To solve this problem, i think we should remove reuse object of RawFormatDeserializationSchema.



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