Infinite loop when InterruptException in @KafkaListener

[coolbeevip]

Affects Version(s):2.5.2.RELEASE

🐞 Bug report

I'm not sure if this is a bug or my usage is wrong

CPU 100% when I throw InterruptEdexception in @KafkaListener

I found an infinite loop in the following code position

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Line 1004 in 1974823

handleConsumerException(e);

In order to reproduce the problem, I throw an exception, all use default parameters

  @KafkaListener(topics = KafkaTopicConfig.TOPIC_NAME, groupId = GROUP_NAME)
  public void listen(String message, Consumer consumer) {
    receivedMessage.add(message);
    if(message.equalsIgnoreCase("test-interrupt")){
      Thread.currentThread().interrupt();
    }
  }

When a message is equal test-interrupt , CPU 100%, and a lot of logs are output as follows

[nc-examples-kafka-biz#0.0.0.0#6065] [000] 2021-06-28 23:24:39,373 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] R [o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer] LogAccessor.java:149 - Consumer exception
java.lang.IllegalStateException: This error handler cannot process 'org.apache.kafka.common.errors.InterruptException's; no record information is available
	at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:151)
	at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:103)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1247)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1004)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:236)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:463)
	at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1275)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1241)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1091)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1047)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:972)
	... 3 common frames omitted
Caused by: java.lang.InterruptedException: null
	... 15 common frames omitted
[nc-examples-kafka-biz#0.0.0.0#6065] [000] 2021-06-28 23:24:39,376 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] R [o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer] LogAccessor.java:149 - Consumer exception
java.lang.IllegalStateException: This error handler cannot process 'org.apache.kafka.common.errors.InterruptException's; no record information is available
	at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:151)
	at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:103)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1247)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1004)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:236)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:463)
	at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1275)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1241)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1091)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1047)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:972)
	... 3 common frames omitted
Caused by: java.lang.InterruptedException: null
	... 15 common frames omitted

[garyrussell]

Use Discussions or stack overflow to ask questions, not issues.

Exactly what are you trying to achieve by interrupting the consumer thread?

[garyrussell]

What you actually do depends on how you want to proceed after the interrupt.

Either way, yes, you need to clear the interrupted flag.

[coolbeevip]

Oh, But I use the third party library catches and swallows InterruptedException, like this

try {
        xxx
    } catch (InterruptedException ie) {
        log.error("xxx",ie);
        Thread.currentThread().interrupt();
    }

This may be a reference to sonar's repair suggestions "InterruptedException" should not be ignored

[coolbeevip]

What you actually do depends on how you want to proceed after the interrupt.

Either way, yes, you need to clear the interrupted flag.

I got it, Thank you for your help

[garyrussell]

This may be a reference ...

Yes; it's best practice not to eat interrupts.

But, in this case, your listener should do so because the interrupt won't be detected by the consumer thread until the next poll and the standard error handlers don't know what to do with it anyway.

The application needs to decide what to do under this condition. Is it fatal, do we need to reprocess this record or send it somewhere else, or ...

[vision57]

Maybe it's better to always clear thread interrupted status before each pollAndInvoke in the loop.