[
https://issues.apache.org/jira/browse/FLINK-3679?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16414130#comment-16414130
]
ASF GitHub Bot commented on FLINK-3679:
---------------------------------------
Github user InfinitiesLoop commented on the issue:
https://github.com/apache/flink/pull/3314
Sorry for necro'ing this thread, but where does the community land on the
multiple record per kafka payload idea this PR originally intended to solve?
I have this scenario, where a single payload in kafka can represent
hundreds of logical records. It's fine to just flatMap() them out after the
deserialization schema, but that does not let me deal with timestamps and
watermarks correctly. It's possible the source is reading from 2 partitions
that are out of sync with each other, but I can't assign a timestamp and
watermark for a single message that contains many records that might span
multiple timestamps. So I'm just using a timestamp and watermark extractor on
the stream separate from the source, and just hoping that I never have out of
sync partitions. If a solution is still desired I'd love to contribute,
otherwise it looks like I will end up having to write my own custom kafka
source..
> Allow Kafka consumer to skip corrupted messages
> -----------------------------------------------
>
> Key: FLINK-3679
> URL: https://issues.apache.org/jira/browse/FLINK-3679
> Project: Flink
> Issue Type: Bug
> Components: DataStream API, Kafka Connector
> Reporter: Jamie Grier
> Assignee: Haohui Mai
> Priority: Major
> Fix For: 1.3.0
>
>
> There are a couple of issues with the DeserializationSchema API that I think
> should be improved. This request has come to me via an existing Flink user.
> The main issue is simply that the API assumes that there is a one-to-one
> mapping between input and outputs. In reality there are scenarios where one
> input message (say from Kafka) might actually map to zero or more logical
> elements in the pipeline.
> Particularly important here is the case where you receive a message from a
> source (such as Kafka) and say the raw bytes don't deserialize properly.
> Right now the only recourse is to throw IOException and therefore fail the
> job.
> This is definitely not good since bad data is a reality and failing the job
> is not the right option. If the job fails we'll just end up replaying the
> bad data and the whole thing will start again.
> Instead in this case it would be best if the user could just return the empty
> set.
> The other case is where one input message should logically be multiple output
> messages. This case is probably less important since there are other ways to
> do this but in general it might be good to make the
> DeserializationSchema.deserialize() method return a collection rather than a
> single element.
> Maybe we need to support a DeserializationSchema variant that has semantics
> more like that of FlatMap.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
