GH-842: Fix NPEs in KafkaListenerAnnotationBPP (#864)

* GH-842: Fix NPEs in KafkaListenerAnnotationBPP

Fixes #842

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

* * Move null check to the `else` branch

* * Do not use `if (resolved == null)` at all

Author: artembilan

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

@@ -242,7 +242,7 @@ public Object postProcessAfterInitialization(final Object bean, final String bea
 			Class<?> targetClass = AopUtils.getTargetClass(bean);
 			Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);
 			final boolean hasClassLevelListeners = classLevelListeners.size() > 0;
-			final List<Method> multiMethods = new ArrayList<Method>();
+			final List<Method> multiMethods = new ArrayList<>();
 			Map<Method, Set<KafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
 					new MethodIntrospector.MetadataLookup<Set<KafkaListener>>() {
 
@@ -295,7 +295,7 @@ public boolean matches(Method method) {
 	 * AnnotationUtils.getRepeatableAnnotations does not look at interfaces
 	 */
 	private Collection<KafkaListener> findListenerAnnotations(Class<?> clazz) {
-		Set<KafkaListener> listeners = new HashSet<KafkaListener>();
+		Set<KafkaListener> listeners = new HashSet<>();
 		KafkaListener ann = AnnotationUtils.findAnnotation(clazz, KafkaListener.class);
 		if (ann != null) {
 			listeners.add(ann);
@@ -325,12 +325,13 @@ private Set<KafkaListener> findListenerAnnotations(Method method) {
 
 	private void processMultiMethodListeners(Collection<KafkaListener> classLevelListeners, List<Method> multiMethods,
 			Object bean, String beanName) {
-		List<Method> checkedMethods = new ArrayList<Method>();
+
+		List<Method> checkedMethods = new ArrayList<>();
 		for (Method method : multiMethods) {
 			checkedMethods.add(checkProxy(method, bean));
 		}
 		for (KafkaListener classLevelListener : classLevelListeners) {
-			MultiMethodKafkaListenerEndpoint<K, V> endpoint = new MultiMethodKafkaListenerEndpoint<K, V>(checkedMethods,
+			MultiMethodKafkaListenerEndpoint<K, V> endpoint = new MultiMethodKafkaListenerEndpoint<>(checkedMethods,
 					bean);
 			endpoint.setBeanFactory(this.beanFactory);
 			processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);
@@ -339,7 +340,7 @@ private void processMultiMethodListeners(Collection<KafkaListener> classLevelLis
 
 	protected void processKafkaListener(KafkaListener kafkaListener, Method method, Object bean, String beanName) {
 		Method methodToUse = checkProxy(method, bean);
-		MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<K, V>();
+		MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();
 		endpoint.setMethod(methodToUse);
 		endpoint.setBeanFactory(this.beanFactory);
 		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
@@ -469,7 +470,7 @@ private Pattern resolvePattern(KafkaListener kafkaListener) {
 			else if (resolved instanceof String) {
 				pattern = Pattern.compile((String) resolved);
 			}
-			else {
+			else if (resolved != null) {
 				throw new IllegalStateException(
 						"topicPattern must resolve to a Pattern or String, not " + resolved.getClass());
 			}
@@ -610,10 +611,11 @@ private String resolveExpressionAsString(String value, String attribute) {
 		if (resolved instanceof String) {
 			return (String) resolved;
 		}
-		else {
+		else if (resolved != null) {
 			throw new IllegalStateException("The [" + attribute + "] must resolve to a String. "
 					+ "Resolved to [" + resolved.getClass() + "] for [" + value + "]");
 		}
+		return null;
 	}
 
 	private Object resolveExpression(String value) {