org.springframework.kafka.listener.ListenerExecutionFailedException：リスナーメソッドがjava.lang.NullPointerExceptionをスローしました

Question

私のコンシューマークラスInventoryEventReceiverのリスナーメソッドで以下のエラーが発生しました。 NullPointerExceptionが表示される理由がわかりません。私はちょうど2つのInventoryEventオブジェクトを投稿しています。

迅速なヘルプがあれば幸いです。私はカフカRESTプロキシ

を介して2つのレコードの下にPOSTしていたリスナー方法

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   
    } 
} 

と

マイConsumerクラス

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が依存関係を管理したい場合（プロセスautowired）、独自にBeanを作成しないでください。

このコードでは、今

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

webhookRepositoryがnullで、設定されていなかったんとして、あなたはNPEを得ました。

InventoryEventReceiverクラスをインスタンス化してインスタンスをwebhookRepositoryにし、作成しないようにする必要があります。