Throw clearer exceptions on serialisation errors
Improved the exception messages that are thrown to indicate whether it was a key or value conversion problem.
Author: Mario Molina
Reviewer: Randall Hauch
@@ -489,10 +489,10 @@ class WorkerSinkTask extends WorkerTask {
}
private SinkRecord convertAndTransformRecord(final ConsumerRecord<byte[], byte[]> msg) {
- SchemaAndValue keyAndSchema = retryWithToleranceOperator.execute(() -> keyConverter.toConnectData(msg.topic(), msg.headers(), msg.key()),
+ SchemaAndValue keyAndSchema = retryWithToleranceOperator.execute(() -> convertKey(msg),
Stage.KEY_CONVERTER, keyConverter.getClass());
- SchemaAndValue valueAndSchema = retryWithToleranceOperator.execute(() -> valueConverter.toConnectData(msg.topic(), msg.headers(), msg.value()),
+ SchemaAndValue valueAndSchema = retryWithToleranceOperator.execute(() -> convertValue(msg),
Stage.VALUE_CONVERTER, valueConverter.getClass());
Headers headers = retryWithToleranceOperator.execute(() -> convertHeadersFor(msg), Stage.HEADER_CONVERTER, headerConverter.getClass());
@@ -524,6 +524,26 @@ class WorkerSinkTask extends WorkerTask {
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.getMessage(), 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.getMessage(), e);
+ throw e;
+ }
+ }
+
private Headers convertHeadersFor(ConsumerRecord<byte[], byte[]> record) {
Headers result = new ConnectHeaders();
org.apache.kafka.common.header.Headers recordHeaders = record.headers();
