NullPointerException in AbstractKafkaDeserializer when using serdes 2.2.5

[ajborley]

Line 119 of the AbstractKafkaDeserializer class is failing with an NPE when running a Kafka consumer client with the "apicurio.registry.headers.enabled" property set to false.

I think the Kafka headers object at line 118 is empty, but not null, meaning that the code attempts to access the headersHandler object at line 119 which has never been initialized in this code block in the AbstractKafkaSerDe class due to the headers.enabled property.

The NPE stack is:

org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition testtopich-0 at offset 29. If needed, please seek past the record to continue consumption.
Caused by: java.lang.NullPointerException
	at io.apicurio.registry.serde.AbstractKafkaDeserializer.deserialize(AbstractKafkaDeserializer.java:119)
	at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1386)
	at org.apache.kafka.clients.consumer.internals.Fetcher.access$3400(Fetcher.java:133)
	at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1617)
	at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1700(Fetcher.java:1453)
	at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:686)
	at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:637)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1303)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1237)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210)

And my consumer client properties are:

props.put(SerdeConfig.ENABLE_HEADERS, false);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, AvroKafkaDeserializer.class);
props.put(AvroKafkaSerdeConfig.USE_SPECIFIC_AVRO_READER, true);