java - Kafka 主题消费者服务 - 侦听器方法抛出异常 java.lang.ArrayIndexOutOfBoundsException

Question

我有一个订阅 Kafka 主题的基于 Spring 的服务。服务工作正常，但有时服务停止工作并出现以下错误。在这种错误情况下，服务停止订阅主题并抛出以下错误。如果我给新的消费者 GroupID，那么服务会再次正常工作。我正在使用 Kafka 进行手动确认。请让我知道可能是什么问题。为什么它需要新的消费者 groupID？

    Caused by: 
org.springframework.kafka.listener.ListenerExecutionFailedException: 
Listener method 'public void 
(org.springframework.messaging.Message<java.lang.String>,org.springframewo
rk.kafka.support.Acknowledgment) throws 
com.fasterxml.jackson.core.JsonParseException,com.fasterxml.jackson.databi
nd.JsonMappingException,java.io.IOException,java.lang.NumberFormatExceptio
n,java.lang.InterruptedException' threw exception; nested exception is 
java.lang.ArrayIndexOutOfBoundsException; nested exception is 
java.lang.ArrayIndexOutOfBoundsException

我在下面提供了 KafkaConfig 详细信息。
这是KafkaConfig：

   @Bean("kafkaListenerContainerFactory")
   @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
   public ConcurrentKafkaListenerContainerFactory<String, String> listenerFactory() {
       ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
       factory.setConsumerFactory(consumerFactory());
       factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);       
        org.springframework.util.backoff.ExponentialBackOff backOff = new org.springframework.util.backoff.ExponentialBackOff(2000L, 1.5);  
        backOff.setMaxInterval(300000L);    
        factory.setErrorHandler(new SeekToCurrentErrorHandler(999999));       
       return factory;
   }



    @KafkaListener(topics = "${topics.kafkaInTopic}", containerFactory = "kafkaListenerContainerFactory") 
public void listen(Message<String> message, Acknowledgment acknowledgment) throws JsonParseException, IOException {

    try {
        //business logic
        acknowledgment.acknowledge();
       }
       Catch {
          //Error handling
          acknowledgment.acknowledge();

       }
}



    Property file:
    consumer:
      auto-offset-reset: latest
      enable-auto-commit: false
      jaas-config: 'org.apache.kafka.common.security.plain.PlainLoginModule  
      sasl-mechanism: PLAIN
      group-id:  sgroup1

详细日志：

2020-11-02T08:42:55.207-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1311)
2020-11-02T08:42:55.207-05:00 [APP/PROC/WEB/1] [OUT] ... 10 common frames omitted
2020-11-02T08:42:55.207-05:00 [APP/PROC/WEB/1] [OUT] Caused by: java.lang.ArrayIndexOutOfBoundsException: null
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] 2020-11-02 13:42:55.253 ERROR [,,,] 15 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.comp.esb.myService.listener.KafkaConsumerOrder.listen(org.springframework.messaging.Message<java.lang.String>,org.springframework.kafka.support.Acknowledgment) throws com.fasterxml.jackson.core.JsonParseException,com.fasterxml.jackson.databind.JsonMappingException,java.io.IOException,java.lang.NumberFormatException,java.lang.InterruptedException' threw exception; nested exception is java.lang.ArrayIndexOutOfBoundsException; nested exception is java.lang.ArrayIndexOutOfBoundsException
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:124)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1296)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1227)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1198)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1118)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:933)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:749)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:698)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] at java.lang.Thread.run(Thread.java:748)
2020-11-02T08:42:55.253-05:00 [APP/PROC/WEB/1] [OUT] Caused by: org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.comp.esb.myService.listener.KafkaConsumerOrder.listen(org.springframework.messaging.Message<java.lang.String>,org.springframework.kafka.support.Acknowledgment) throws com.fasterxml.jackson.core.JsonParseException,com.fasterxml.jackson.databind.JsonMappingException,java.io.IOException,java.lang.NumberFormatException,java.lang.InterruptedException' threw exception; nested exception is java.lang.ArrayIndexOutOfBoundsException; nested exception is java.lang.ArrayIndexOutOfBoundsException

标签： javaapache-kafkakafka-consumer-apispring-kafka