[FLINK-32303] Incorrect error message in KafkaSource

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 1.18.0, 1.17.1
• Component/s: Connectors / Kafka
• Assignee: Unassigned
• Reporter: Piotr Nowojski
• Created: 09/Jun/23 16:43
• Updated: 09/Jun/23 16:48

Description

When exception is thrown from an operator chained with a KafkaSource, KafkaSource is returning a misleading error, like shown below:

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

Issue Links

• is caused by: FLINK-25132 KafkaSource cannot work with object-reusing DeserializationSchema