fred-ro commented on code in PR #15691:
URL: https://github.com/apache/kafka/pull/15691#discussion_r1617602453
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java:
##########
@@ -311,25 +312,37 @@ <K, V> ConsumerRecord<K, V> parseRecord(Deserializers<K,
V> deserializers,
Optional<Integer> leaderEpoch,
TimestampType timestampType,
Record record) {
+ ByteBuffer keyBytes = record.key();
+ ByteBuffer valueBytes = record.value();
+ Headers headers = new RecordHeaders(record.headers());
+ K key;
+ V value;
try {
- long offset = record.offset();
- long timestamp = record.timestamp();
- Headers headers = new RecordHeaders(record.headers());
- ByteBuffer keyBytes = record.key();
- K key = keyBytes == null ? null :
deserializers.keyDeserializer.deserialize(partition.topic(), headers, keyBytes);
- ByteBuffer valueBytes = record.value();
- V value = valueBytes == null ? null :
deserializers.valueDeserializer.deserialize(partition.topic(), headers,
valueBytes);
- return new ConsumerRecord<>(partition.topic(),
partition.partition(), offset,
- timestamp, timestampType,
- keyBytes == null ? ConsumerRecord.NULL_SIZE :
keyBytes.remaining(),
- valueBytes == null ? ConsumerRecord.NULL_SIZE :
valueBytes.remaining(),
- key, value, headers, leaderEpoch);
+ key = keyBytes == null ? null :
deserializers.keyDeserializer.deserialize(partition.topic(), headers, keyBytes);
} catch (RuntimeException e) {
- log.error("Deserializers with error: {}", deserializers);
Review Comment:
I modified to add different log if error is from key or value deserializer.
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
