[GitHub] [kafka] mmolimar commented on a change in pull request #7496: KAFKA-9018: Throw clearer exceptions on serialisation errors

[GitBox]

mmolimar commented on a change in pull request #7496:
URL: https://github.com/apache/kafka/pull/7496#discussion_r448045788



##########
File path: 
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
##########
@@ -524,6 +524,26 @@ private SinkRecord convertAndTransformRecord(final 
ConsumerRecord<byte[], byte[]
         return new InternalSinkRecord(msg, transformedRecord);
     }
 
+    private SchemaAndValue convertKey(ConsumerRecord<byte[], byte[]> msg) {
+        try {
+            return keyConverter.toConnectData(msg.topic(), msg.headers(), 
msg.key());
+        } catch (Exception e) {
+            log.error("{} Error converting message key in topic '{}' partition 
{} at offset {} and timestamp {}",
+                    this, msg.topic(), msg.partition(), msg.offset(), 
msg.timestamp(), e);
+            throw e;
+        }
+    }
+
+    private SchemaAndValue convertValue(ConsumerRecord<byte[], byte[]> msg) {
+        try {
+            return valueConverter.toConnectData(msg.topic(), msg.headers(), 
msg.value());
+        } catch (Exception e) {
+            log.error("{} Error converting message value in topic '{}' 
partition {} at offset {} and timestamp {}",
+                    this, msg.topic(), msg.partition(), msg.offset(), 
msg.timestamp(), e);

Review comment:
       Done!




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org