Process errorHandler in class level KafkaListener

The `errorHandler` attributed has been missed from the
`KafkaListenerAnnotationBeanPostProcessor.processMultiMethodListeners()`
 logic.

* Move `errorHandler` and `BeanFactory` population into the
`processListener()` method in the `KafkaListenerAnnotationBeanPostProcessor`

**Cherry-pick to 2.1.x, 2.0.x & 1.3.x**

Fix Test

Author: artembilan

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -333,7 +333,6 @@ private void processMultiMethodListeners(Collection<KafkaListener> classLevelLis
 		for (KafkaListener classLevelListener : classLevelListeners) {
 			MultiMethodKafkaListenerEndpoint<K, V> endpoint = new MultiMethodKafkaListenerEndpoint<>(checkedMethods,
 					bean);
-			endpoint.setBeanFactory(this.beanFactory);
 			processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);
 		}
 	}
@@ -342,11 +341,6 @@ protected void processKafkaListener(KafkaListener kafkaListener, Method method,
 		Method methodToUse = checkProxy(method, bean);
 		MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();
 		endpoint.setMethod(methodToUse);
-		endpoint.setBeanFactory(this.beanFactory);
-		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
-		if (StringUtils.hasText(errorHandlerBeanName)) {
-			endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));
-		}
 		processListener(endpoint, kafkaListener, bean, methodToUse, beanName);
 	}
 
@@ -413,6 +407,11 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 			}
 		}
 
+		endpoint.setBeanFactory(this.beanFactory);
+		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
+		if (StringUtils.hasText(errorHandlerBeanName)) {
+			endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));
+		}
 		this.registrar.registerEndpoint(endpoint, factory);
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -255,6 +255,8 @@ public void testMulti() throws Exception {
 		template.flush();
 		assertThat(this.multiListener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.multiListener.latch2.await(60, TimeUnit.SECONDS)).isTrue();
+		template.send("annotated8", 0, 1, "junk");
+		assertThat(this.multiListener.errorLatch.await(60, TimeUnit.SECONDS)).isTrue();
 	}
 
 	@Test
@@ -607,6 +609,14 @@ public KafkaListenerErrorHandler consumeException(Listener listener) {
 			};
 		}
 
+		@Bean
+		public KafkaListenerErrorHandler consumeMultiMethodException(MultiListenerBean listener) {
+			return (m, e) -> {
+				listener.errorLatch.countDown();
+				return null;
+			};
+		}
+
 	}
 
 	static class Listener implements ConsumerSeekAware {
@@ -864,16 +874,23 @@ public CountDownLatch getLatch2() {
 
 	}
 
-	@KafkaListener(id = "multi", topics = "annotated8")
+	@KafkaListener(id = "multi", topics = "annotated8", errorHandler = "consumeMultiMethodException")
 	static class MultiListenerBean {
 
 		private final CountDownLatch latch1 = new CountDownLatch(1);
 
 		private final CountDownLatch latch2 = new CountDownLatch(1);
 
+		private final CountDownLatch errorLatch = new CountDownLatch(1);
+
 		@KafkaHandler
 		public void bar(@Payload String bar) {
-			latch1.countDown();
+			if ("junk".equals(bar)) {
+				throw new RuntimeException("intentional");
+			}
+			else {
+				this.latch1.countDown();
+			}
 		}
 
 		@KafkaHandler