[ https://issues.apache.org/jira/browse/KAFKA-7036?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Chia-Ping Tsai resolved KAFKA-7036. ----------------------------------- Resolution: Won't Fix > Complete the docs of KafkaConsumer#poll > --------------------------------------- > > Key: KAFKA-7036 > URL: https://issues.apache.org/jira/browse/KAFKA-7036 > Project: Kafka > Issue Type: Improvement > Reporter: Chia-Ping Tsai > Assignee: Chia-Ping Tsai > Priority: Minor > > KafkaConsumer#poll has a nice docs about the expected exceptions. However, it > lacks the description of SerializationException. Another mirror issue is that > KafkaConsumer doesn't catch all type of exception which may be thrown by > deserializer (see below). We should use Throwable to replace the > RuntimeException so as to catch all exception and then wrap them to > SerializationException. > {code:java} > private ConsumerRecord<K, V> parseRecord(TopicPartition partition, > RecordBatch batch, > Record record) { > try { > long offset = record.offset(); > long timestamp = record.timestamp(); > TimestampType timestampType = batch.timestampType(); > Headers headers = new RecordHeaders(record.headers()); > ByteBuffer keyBytes = record.key(); > byte[] keyByteArray = keyBytes == null ? null : > Utils.toArray(keyBytes); > K key = keyBytes == null ? null : > this.keyDeserializer.deserialize(partition.topic(), headers, keyByteArray); > ByteBuffer valueBytes = record.value(); > byte[] valueByteArray = valueBytes == null ? null : > Utils.toArray(valueBytes); > V value = valueBytes == null ? null : > this.valueDeserializer.deserialize(partition.topic(), headers, > valueByteArray); > return new ConsumerRecord<>(partition.topic(), partition.partition(), > offset, > timestamp, timestampType, > record.checksumOrNull(), > keyByteArray == null ? > ConsumerRecord.NULL_SIZE : keyByteArray.length, > valueByteArray == null ? > ConsumerRecord.NULL_SIZE : valueByteArray.length, > key, value, headers); > } catch (RuntimeException e) { > throw new SerializationException("Error deserializing key/value for > partition " + partition + > " at offset " + record.offset() + ". If needed, please seek > past the record to continue consumption.", e); > } > }{code} -- This message was sent by Atlassian Jira (v8.3.4#803005)