diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 3b88b466..5c15af8f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -1256,7 +1256,8 @@ public class KafkaMessageListenerContainer // NOSONAR line count } else if (this.isBatchListener && this.batchErrorHandler != null) { this.batchErrorHandler.handle(e, new ConsumerRecords(Collections.emptyMap()), this.consumer, - KafkaMessageListenerContainer.this.thisOrParentContainer); + KafkaMessageListenerContainer.this.thisOrParentContainer, () -> { + }); } else { this.logger.error(e, "Consumer exception"); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java index 7c68b259..6b701be7 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java @@ -83,6 +83,10 @@ public class RetryingBatchErrorHandler implements ListenerInvokingBatchErrorHand public void handle(Exception thrownException, ConsumerRecords records, Consumer consumer, MessageListenerContainer container, Runnable invokeListener) { + if (records.count() == 0) { + LOGGER.error(thrownException, "Called with no records; consumer exception"); + return; + } BackOffExecution execution = this.backOff.start(); long nextBackOff = execution.nextBackOff(); String failed = null; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerIntegrationTests.java index 846c0702..dbe907a9 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/RetryingBatchErrorHandlerIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,10 @@ package org.springframework.kafka.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; import java.util.List; import java.util.Map; @@ -28,14 +32,17 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.event.ConsumerStoppedEvent; +import org.springframework.kafka.support.TopicPartitionOffset; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; import org.springframework.kafka.test.context.EmbeddedKafka; @@ -209,4 +216,31 @@ public class RetryingBatchErrorHandlerIntegrationTests { assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue(); } + @SuppressWarnings("unchecked") + @Test + void consumerEx() throws InterruptedException { + ConsumerFactory cf = mock(ConsumerFactory.class); + Consumer consumer = mock(Consumer.class); + given(consumer.poll(any())).willThrow(new RuntimeException("test")); + given(cf.createConsumer(any(), any(), isNull(), any())).willReturn(consumer); + ContainerProperties containerProps = new ContainerProperties(new TopicPartitionOffset("foo", 0)); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, + containerProps); + CountDownLatch called = new CountDownLatch(1); + container.setBatchErrorHandler(new RetryingBatchErrorHandler() { + + @Override + public void handle(Exception thrownException, ConsumerRecords records, Consumer consumer, + MessageListenerContainer container, Runnable invokeListener) { + + called.countDown(); + super.handle(thrownException, records, consumer, container, invokeListener); + } + }); + container.setupMessageListener((BatchMessageListener) (recs -> { })); + container.start(); + assertThat(called.await(10, TimeUnit.SECONDS)).isTrue(); + container.stop(); + } + }