[NIFI-4046] ConsumeKafkaRecord can throw NPE if using an Avro Reader and data on topic is missing a required field

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.4.0
• Component/s: Extensions
• Assignee: Mark Payne
• Reporter: Mark Payne
• Created: 08/Jun/17 19:07
• Updated: 30/Jun/17 12:57
• Resolved: 30/Jun/17 12:57

Description

I am using PublishKafkaRecord to push some data to a Kafka topic using the Avro writer. I not embedding the Avro schema in the content but rather using a Schema Registry.

I then have a ConsumeKafkaRecord that pulls that Avro data, but is using a Schema that has an extra field, which is not present in the data. As a result, I get a NullPointerException when trying to pull data from Kafka:

2017-06-08 14:54:12,499 ERROR [Timer-Driven Process Thread-22] o.a.n.p.k.pubsub.ConsumeKafkaRecord_0_10 ConsumeKafkaRecord_0_10[id=84480f03-015c-1000-4aab-9a035812bb86] Exception while processing data from kafka so will close the lease org.apache.nifi.processors.kafka.pubsub.ConsumerPool$SimpleConsumerLease@6c39d011 due to org.apache.nifi.processor.exception.ProcessException: java.lang.NullPointerException: org.apache.nifi.processor.exception.ProcessException: java.lang.NullPointerException
org.apache.nifi.processor.exception.ProcessException: java.lang.NullPointerException
        at org.apache.nifi.processors.kafka.pubsub.ConsumerLease.writeRecordData(ConsumerLease.java:514)
        at org.apache.nifi.processors.kafka.pubsub.ConsumerLease.lambda$processRecords$2(ConsumerLease.java:320)
        at java.util.HashMap$KeySpliterator.forEachRemaining(HashMap.java:1540)
        at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
        at org.apache.nifi.processors.kafka.pubsub.ConsumerLease.processRecords(ConsumerLease.java:307)
        at org.apache.nifi.processors.kafka.pubsub.ConsumerLease.poll(ConsumerLease.java:168)
        at org.apache.nifi.processors.kafka.pubsub.ConsumeKafkaRecord_0_10.onTrigger(ConsumeKafkaRecord_0_10.java:327)
        at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27)
        at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1120)
        at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:147)
        at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:47)
        at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:132)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.NullPointerException: null
        at org.apache.nifi.processors.kafka.pubsub.ConsumerLease.writeRecordData(ConsumerLease.java:458)
        ... 18 common frames omitted

Issue Links

• links to: GitHub Pull Request #1906