[
https://issues.apache.org/jira/browse/FLINK-32303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17731021#comment-17731021
]
Piotr Nowojski commented on FLINK-32303:
----------------------------------------
I think this bug has been introduced by FLINK-25132 and I think a solution
should be to just simply remove the
{code:java}
try {
} catch (Exception e) {
throw new IOException("Failed to deserialize consumer record due
to", e);
}
{code}
section in {{KafkaRecordEmitter#emitRecord}}. [~renqs], WDYT?
> Incorrect error message in KafkaSource
> ---------------------------------------
>
> Key: FLINK-32303
> URL: https://issues.apache.org/jira/browse/FLINK-32303
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 1.18.0, 1.17.1
> Reporter: Piotr Nowojski
> Priority: Minor
>
> When exception is thrown from an operator chained with a KafkaSource,
> KafkaSource is returning a misleading error, like shown below:
> {noformat}
> java.io.IOException: Failed to deserialize consumer record due to
> at
> org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:56)
> ~[classes/:?]
> at
> org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:33)
> ~[classes/:?]
> at
> org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:160)
> ~[classes/:?]
> at
> org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:417)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:562)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:852)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:801)
> ~[classes/:?]
> at
> org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
> ~[classes/:?]
> at
> org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931)
> [classes/:?]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745)
> [classes/:?]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
> [classes/:?]
> at java.lang.Thread.run(Thread.java:829) [?:?]
> Caused by:
> org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException:
> Could not forward element to next operator
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:92)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:309)
> ~[classes/:?]
> at
> org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
> ~[classes/:?]
> at
> org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter$SourceOutputWrapper.collect(KafkaRecordEmitter.java:67)
> ~[classes/:?]
> at
> org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:84)
> ~[classes/:?]
> at
> org.apache.flink.connector.kafka.source.reader.deserializer.KafkaValueOnlyDeserializationSchemaWrapper.deserialize(KafkaValueOnlyDeserializationSchemaWrapper.java:51)
> ~[classes/:?]
> at
> org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53)
> ~[classes/:?]
> ... 14 more
> Caused by:
> org.apache.flink.runtime.operators.testutils.ExpectedTestException: Failover!
> at
> org.apache.flink.test.ManySmallJobsBenchmarkITCase$ThrottlingAndFailingIdentityMap.map(ManySmallJobsBenchmarkITCase.java:263)
> ~[test-classes/:?]
> at
> org.apache.flink.test.ManySmallJobsBenchmarkITCase$ThrottlingAndFailingIdentityMap.map(ManySmallJobsBenchmarkITCase.java:243)
> ~[test-classes/:?]
> at
> org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:38)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
> ~[classes/:?]
> at
> org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:309)
> ~[classes/:?]
> at
> org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
> ~[classes/:?]
> at
> org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter$SourceOutputWrapper.collect(KafkaRecordEmitter.java:67)
> ~[classes/:?]
> at
> org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:84)
> ~[classes/:?]
> at
> org.apache.flink.connector.kafka.source.reader.deserializer.KafkaValueOnlyDeserializationSchemaWrapper.deserialize(KafkaValueOnlyDeserializationSchemaWrapper.java:51)
> ~[classes/:?]
> at
> org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53)
> ~[classes/:?]
> ... 14 more
> {noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
