[jira] [Updated] (FLINK-39320) Include Kafka record metadata in deserialization error message

["Pritesh Gudge (Jira)" <ji...@apache.org>]

     [ 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 ~6-line 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:

{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 in KafkaRecordEmitter.java 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 ~6-line 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)