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**

Author: artembilan

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

@@ -325,7 +325,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);
 		}
 	}
@@ -334,11 +333,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);
 	}
 
@@ -405,6 +399,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

@@ -281,6 +281,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
@@ -1273,18 +1275,33 @@ public CountDownLatch getLatch2() {
 			return latch2;
 		}
 
+		@Bean
+		public KafkaListenerErrorHandler consumeMultiMethodException(MultiListenerBean listener) {
+			return (m, e) -> {
+				listener.errorLatch.countDown();
+				return null;
+			};
+		}
+
 	}
 
-	@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(@NonNull String bar) {
-			latch1.countDown();
+			if ("junk".equals(bar)) {
+				throw new RuntimeException("intentional");
+			}
+			else {
+				this.latch1.countDown();
+			}
 		}
 
 		@KafkaHandler