GH-667: Fix BatchErrorHandler for generic errors

Fixes #667

When catching exceptions outside of the listener (e.g. commit errors), the
`BatchErrorHandler` was never called since the record error handler gets a
logging handler.

Check listener type before invoking the error handler for consumer errors.

Author: garyrussell

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

@@ -781,11 +781,11 @@ public void run() {
 		 */
 		protected void handleConsumerException(Exception e) {
 			try {
-				if (this.errorHandler != null) {
+				if (!this.isBatchListener && this.errorHandler != null) {
 					this.errorHandler.handle(e, Collections.emptyList(), this.consumer,
 							KafkaMessageListenerContainer.this);
 				}
-				else if (this.batchErrorHandler != null) {
+				else if (this.isBatchListener && this.batchErrorHandler != null) {
 					this.batchErrorHandler.handle(e, new ConsumerRecords<K, V>(Collections.emptyMap()), this.consumer,
 							KafkaMessageListenerContainer.this);
 				}

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -1968,6 +1968,54 @@ public void testAckModeCount() throws Exception {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testCommitErrorHandlerCalled() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
+		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
+		records.put(new TopicPartition("foo", 0), Arrays.asList(
+				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
+				new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
+		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		ConsumerRecords<Integer, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
+		AtomicBoolean first = new AtomicBoolean(true);
+		given(consumer.poll(anyLong())).willAnswer(i -> {
+			Thread.sleep(50);
+			return first.getAndSet(false) ? consumerRecords : emptyRecords;
+		});
+		willAnswer(i -> {
+			throw new RuntimeException("Commit failed");
+		}).given(consumer).commitSync(any(Map.class));
+		TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] {
+				new TopicPartitionInitialOffset("foo", 0) };
+		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+		containerProps.setGroupId("grp");
+		containerProps.setClientId("clientId");
+		containerProps.setIdleEventInterval(100L);
+		containerProps.setMessageListener((MessageListener) r -> { });
+		final CountDownLatch ehl = new CountDownLatch(1);
+		containerProps.setErrorHandler((r, t) -> {
+			ehl.countDown();
+		});
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.start();
+		assertThat(ehl.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		containerProps.setMessageListener((BatchMessageListener) r -> { });
+		final CountDownLatch behl = new CountDownLatch(1);
+		containerProps.setBatchErrorHandler((r, t) -> {
+			behl.countDown();
+		});
+		container = new KafkaMessageListenerContainer<>(cf, containerProps);
+		first.set(true);
+		container.start();
+		assertThat(behl.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+	}
+
 	private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
 		Consumer<?, ?> consumer = spy(
 				KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class));