[
https://issues.apache.org/jira/browse/FLINK-26018?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17490823#comment-17490823
]
Qingsheng Ren commented on FLINK-26018:
---------------------------------------
This issue exists for all sources extending SourceReaderBase, so I think
KafkaSource in 1.13 also has this problem.
A solution in my mind is that we can create per-split output for a split once
it's assigned to the source operator. For splits assigned before the main
reader output is created (before the first {{SourceOperator#emitNext}}
invocation), we can use a list to hold them temporarily in SourceOperator and
create per-split outputs when the main output is ready.
> Unnecessary late events when using the new KafkaSource
> ------------------------------------------------------
>
> Key: FLINK-26018
> URL: https://issues.apache.org/jira/browse/FLINK-26018
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 1.14.3
> Reporter: Jun Qin
> Assignee: Qingsheng Ren
> Priority: Critical
> Attachments: message in kafka.txt,
> taskmanager_10.28.0.131_33249-b3370c_log
>
>
> There is an issue with the new KafkaSource connector in Flink 1.14: when one
> task consumes messages from multiple topic partitions (statically created,
> timestamp are in order), it may start with one partition and advances
> watermarks before the data from other partitions come. In this case, the
> early messages in other partitions may unnecessarily be considered as late
> ones.
> I discussed with [~renqs], it seems that the new KafkaSource only adds a
> partition into {{WatermarkMultiplexer}} when it receives data from that
> partition. In contrast, FlinkKafkaConsumer adds all known partition before it
> fetch any data.
> Attached two files: the messages in Kafka and the corresponding TM logs.
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
