GH-1855: Fix Call to RetryingBatchErrorHandler

Resolves https://github.com/spring-projects/spring-kafka/issues/1855

Unsupported op when a consumer exception occurs.

**cherry-pick to all 2.x.x (down to and including 2.3.x)**

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

@@ -1256,7 +1256,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 				}
 				else if (this.isBatchListener && this.batchErrorHandler != null) {
 					this.batchErrorHandler.handle(e, new ConsumerRecords<K, V>(Collections.emptyMap()), this.consumer,
-							KafkaMessageListenerContainer.this.thisOrParentContainer);
+							KafkaMessageListenerContainer.this.thisOrParentContainer, () -> {
+							});
 				}
 				else {
 					this.logger.error(e, "Consumer exception");

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;

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<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> 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<Integer, String> 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<Integer, String>) (recs -> { }));
+		container.start();
+		assertThat(called.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+	}
+
 }