org.springframework.kafka.listener.ListenerExecutionFailedException：监听器方法抛出了java.lang.NullPointerException

Question

我在侦听器方法中的Consumer类InventoryEventReceiver中收到以下错误。不确定NullPointerException出现的原因。我只是张贴两个InventoryEvent对象。

任何快速帮助将不胜感激。

我的Consumer类有监听器方法

public class InventoryEventReceiver {

    private static final Logger log = LoggerFactory.getLogger(InventoryEventReceiver.class);

    private CountDownLatch latch = new CountDownLatch(1);

    public CountDownLatch getLatch() {
        return latch;
    }

    @KafkaListener(topics="inventory", containerFactory="kafkaListenerContainerFactory")
    public void listenWithHeaders(
            InventoryEvent event) {

        System.out.println("EVENT HAS BEEN RECEIVED by listenWithHeaders(InventoryEvent)");
        System.out.println(event.toString());


        log.info(System.currentTimeMillis() + "-- Received Event :\"" + event + " for topic : inventory");
        System.out.println("Sending event to webhook triggers ... ");
        KafkaWebhookServiceImpl webhookService = new KafkaWebhookServiceImpl();

        List<WebhookRequestBody> listWebhooks = webhookService.getAllWebhooksForTopic("inventory");
        System.out.println("Number of registered webhooks for topic \"inventory\" : " + listWebhooks.size());

        CountDownLatch countLatch = new CountDownLatch(listWebhooks.size());

        for(WebhookRequestBody w : listWebhooks) {
            Executors.newSingleThreadExecutor().execute(new InventoryEventProcessor(countLatch, event, w));
        }

        try {
            countLatch.await();  // wait until countLatch counted down to 0
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Events SENT to all listening webhook triggers. ");

        latch.countDown();
    }
}

KafkaWebhookServiceImpl类

@Service("webhookService")
@Transactional
public class KafkaWebhookServiceImpl implements KafkaWebhookService {

    @Autowired
    private KafkaWebhookRepository webhookRepository;

    @Override
    public List<WebhookRequestBody> getAllWebhooksForTopic(String topic) {
        return webhookRepository.findByTopic(topic); <-- ERROR: line 45         
    }
}

我通过Kafka REST Proxy发布以下两条记录

curl -i -X POST -H "Content-Type: application/vnd.kafka.json.v1+json" --data '{"value_schema": "{\"type\": \"record\", \"name\": \"InventoryEvent\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"},{\"name\": \"eventType\", \"type\": \"string\"},{\"name\": \"qtyReq\", \"type\": \"int\"},{\"name\": \"qtyLevel\", \"type\": \"int\"}]}", "records": [{"value": {"id": 6122,"eventType":"inventory.transaction","qtyReq": 34,"qtyLevel": 129}},{"value": {"id": 7798,"eventType":"inventory.transaction","qtyReq": 5,"qtyLevel": 27}}]}' http://localhost:8082/topics/inventory

错误日志

EVENT HAS BEEN RECEIVED by listenWithHeaders(InventoryEvent)
InventoryEvent [id=7798, eventType='inventory.transaction', qtyReq='5', qtyLevel='27']
2017-12-29 10:51:22.375  INFO 12418 --- [ntainer#0-0-C-1] c.p.kafka.spring.InventoryEventReceiver  : 1514544682375-- Received Event :"InventoryEvent [id=7798, eventType='inventory.transaction', qtyReq='5', qtyLevel='27'] for topic : inventory
Sending event to webhook triggers ...
2017-12-29 10:51:22.376 ERROR 12418 --- [ntainer#0-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: ConsumerRecord(topic = inventory, partition = 0, offset = 23, CreateTime = 1514544682080, checksum = 1801448922, serialized key size = -1, serialized value size = 72, key = null, value = InventoryEvent [id=7798, eventType='inventory.transaction', qtyReq='5', qtyLevel='27'])

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.psl.kafka.spring.InventoryEventReceiver.listenWithHeaders(com.psl.kafka.spring.InventoryEvent)' threw exception; nested exception is java.lang.NullPointerException
        at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:188) ~[spring-kafka-1.1.7.RELEASE.jar:na]
        at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:72) ~[spring-kafka-1.1.7.RELEASE.jar:na]
        at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:47) ~[spring-kafka-1.1.7.RELEASE.jar:na]
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:792) [spring-kafka-1.1.7.RELEASE.jar:na]
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:736) [spring-kafka-1.1.7.RELEASE.jar:na]
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:568) [spring-kafka-1.1.7.RELEASE.jar:na]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_151]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_151]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_151]
Caused by: java.lang.NullPointerException: null
        at com.psl.kafka.rest.KafkaWebhookServiceImpl.getAllWebhooksForTopic(KafkaWebhookServiceImpl.java:45) ~[classes/:0.0.1-SNAPSHOT]
        at com.psl.kafka.spring.InventoryEventReceiver.listenWithHeaders(InventoryEventReceiver.java:126) ~[classes/:0.0.1-SNAPSHOT]
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_151]
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_151]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_151]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_151]
        at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:180) ~[spring-messaging-4.3.13.RELEASE.jar:4.3.13.RELEASE]
        at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:112) ~[spring-messaging-4.3.13.RELEASE.jar:4.3.13.RELEASE]
        at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:48) ~[spring-kafka-1.1.7.RELEASE.jar:na]
        at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:174) ~[spring-kafka-1.1.7.RELEASE.jar:na]
        ... 8 common frames omitted

Answer 1

问题出在InventoryEventReceiver类的KafkaWebhookServiceImpl webhookService = new KafkaWebhookServiceImpl();中。

如果你想让spring管理依赖项（进程自动连接），你不应该自己创建bean。

现在在这段代码中

@Override
public List<WebhookRequestBody> getAllWebhooksForTopic(String topic) {
    return webhookRepository.findByTopic(topic); <-- ERROR: line 45         
}

你有NPE，因为webhookRepository是null并且从未设置过。

您需要重写InventoryEventReceiver类以获取webhookRepository的实例，而不是创建它。