[
https://issues.apache.org/jira/browse/FLINK-39320?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Pritesh Gudge updated FLINK-39320:
----------------------------------
Description:
When deserialization fails in KafkaRecordEmitter, the error message doesn't
include which Kafka record caused the failure:
{code:java}
throw new IOException("Failed to deserialize consumer record due to", e);
{code}
In production environments with many topics and partitions, this makes
root-cause analysis difficult. Engineers have to add custom logging and
redeploy to locate the problematic record.
Proposed change: include topic, partition, and offset in the exception
message in
[KafkaRecordEmitter.java|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaRecordEmitter.java#L55-L56]:
{code:java}
throw new IOException(
String.format(
"Failed to deserialize consumer record from topic=%s, partition=%d,
offset=%d due to",
consumerRecord.topic(),
consumerRecord.partition(),
consumerRecord.offset()),
e);
{code}
This is a logging change with no API or behavioral impact. The ConsumerRecord
already has this metadata available.
was:
When deserialization fails in KafkaRecordEmitter, the error message doesn't
include which Kafka record caused the failure:
{code:java}
throw new IOException("Failed to deserialize consumer record due to", e);
{code}
In production environments with many topics and partitions, this makes
root-cause analysis difficult. Engineers have to add custom logging and
redeploy to locate the problematic record.
Proposed change: include topic, partition, and offset in the exception
message in
[KafkaRecordEmitter.java|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaRecordEmitter.java#L55-L56]:
{code:java}
throw new IOException(
String.format(
"Failed to deserialize consumer record from topic=%s, partition=%d,
offset=%d due to",
consumerRecord.topic(),
consumerRecord.partition(),
consumerRecord.offset()),
e);
{code}
This is a ~6-line change with no API or behavioral impact. The ConsumerRecord
already has this metadata available.
> Include Kafka record metadata in deserialization error message
> --------------------------------------------------------------
>
> Key: FLINK-39320
> URL: https://issues.apache.org/jira/browse/FLINK-39320
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Kafka
> Reporter: Pritesh Gudge
> Priority: Minor
>
> When deserialization fails in KafkaRecordEmitter, the error message doesn't
> include which Kafka record caused the failure:
> {code:java}
> throw new IOException("Failed to deserialize consumer record due to", e);
> {code}
> In production environments with many topics and partitions, this makes
> root-cause analysis difficult. Engineers have to add custom logging and
> redeploy to locate the problematic record.
> Proposed change: include topic, partition, and offset in the exception
> message in
>
> [KafkaRecordEmitter.java|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaRecordEmitter.java#L55-L56]:
> {code:java}
> throw new IOException(
> String.format(
> "Failed to deserialize consumer record from topic=%s, partition=%d,
> offset=%d due to",
> consumerRecord.topic(),
> consumerRecord.partition(),
> consumerRecord.offset()),
> e);
> {code}
> This is a logging change with no API or behavioral impact. The
> ConsumerRecord already has this metadata available.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
