[LI-HOTFIX] Add serialization error metric in record-errors (#437) (#441

) TICKET = LIKAFKA-49998 LI_DESCRIPTION = Add producer metric for serialization error EXIT_CRITERIA = N/A Co-authored-by: Qi Liu <[email protected]>
linkedin · Feb 22, 2023 · d8e93c3 · d8e93c3
1 parent dec729a
commit d8e93c3
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 0 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -954,6 +954,7 @@ private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback call
             try {
                 serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
             } catch (ClassCastException cce) {
+                sender.sendError(record.topic(), 1);
                 throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() +
                         " to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +
                         " specified in key.serializer", cce);
@@ -962,6 +963,7 @@ private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback call
             try {
                 serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());
             } catch (ClassCastException cce) {
+                sender.sendError(record.topic(), 1);
                 throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() +
                         " to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +
                         " specified in value.serializer", cce);

diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
@@ -846,6 +846,13 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
         log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
     }
 
+    /**
+     * Send record error as error metric to sensor
+     */
+    public void sendError(String topic, int count) {
+        this.sensors.recordErrors(topic, count);
+    }
+
     /**
      * Wake up the selector associated with this send thread
      */