GH-950: Always wrap listener exception in LEFE

Fixes #950

Provide access to the `group.id` for all listener types, including
simple listeners.

* Fix tests and KafkaStreams.close() deprecation.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2018 the original author or authors.
+ * Copyright 2017-2019 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.config;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
@@ -60,7 +61,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private static final String CLEANUP_CONFIG_MUST_NOT_BE_NULL = "'cleanupConfig' must not be null";
 
-	private static final int DEFAULT_CLOSE_TIMEOUT = 10;
+	private static final Duration DEFAULT_CLOSE_TIMEOUT = Duration.ofSeconds(10);
 
 	private KafkaClientSupplier clientSupplier = new DefaultKafkaClientSupplier();
 
@@ -82,7 +83,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private int phase = Integer.MAX_VALUE - 1000; // NOSONAR magic #
 
-	private int closeTimeout = DEFAULT_CLOSE_TIMEOUT;
+	private Duration closeTimeout = DEFAULT_CLOSE_TIMEOUT;
 
 	private KafkaStreams kafkaStreams;
 
@@ -241,7 +242,7 @@ public void setStateRestoreListener(StateRestoreListener stateRestoreListener) {
 	 * @see KafkaStreams#close(long, TimeUnit)
 	 */
 	public void setCloseTimeout(int closeTimeout) {
-		this.closeTimeout = closeTimeout; // NOSONAR (sync)
+		this.closeTimeout = Duration.ofSeconds(closeTimeout); // NOSONAR (sync)
 	}
 
 	@Override
@@ -319,7 +320,7 @@ public synchronized void stop() {
 		if (this.running) {
 			try {
 				if (this.kafkaStreams != null) {
-					this.kafkaStreams.close(this.closeTimeout, TimeUnit.SECONDS);
+					this.kafkaStreams.close(this.closeTimeout);
 					if (this.cleanupConfig.cleanupOnStop()) {
 						this.kafkaStreams.cleanUp();
 					}

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

@@ -112,7 +112,7 @@
  * @author Yang Qiju
  * @author Tom van den Berge
  */
-public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
+public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		extends AbstractMessageListenerContainer<K, V> {
 
 	private static final int DEFAULT_ACK_TIME = 5000;
@@ -1271,13 +1271,9 @@ private Exception decorateException(RuntimeException e) {
 				toHandle = new ListenerExecutionFailedException(toHandle.getMessage(), this.consumerGroupId,
 						toHandle.getCause());
 			}
-//			else {
-//				/*
-//				 * TODO: in 2.3, wrap all exceptions (e.g. thrown by user implementations
-//				 * of MessageListener) in LEFE with groupId. @KafkaListeners always throw
-//				 * LEFE.
-//				 */
-//			}
+			else {
+				toHandle = new ListenerExecutionFailedException("Listener failed", this.consumerGroupId, toHandle);
+			}
 			return toHandle;
 		}
 

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

@@ -155,11 +155,11 @@ private ConcurrentKafkaListenerContainerFactory<String, String> factory(Consumer
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(cf);
 			factory.setErrorHandler((t, r) -> {
-				if (r.value() == null && t instanceof DeserializationException) {
+				if (r.value() == null && t.getCause() instanceof DeserializationException) {
 					this.valueErrorCount.incrementAndGet();
-					this.headers = ((DeserializationException) t).getHeaders();
+					this.headers = ((DeserializationException) t.getCause()).getHeaders();
 				}
-				else if (r.key() == null && t instanceof DeserializationException) {
+				else if (r.key() == null && t.getCause() instanceof DeserializationException) {
 					this.keyErrorCount.incrementAndGet();
 				}
 				this.latch.countDown();

src/reference/asciidoc/whats-new.adoc

@@ -7,4 +7,8 @@ This section covers the changes made from version 2.2 to version 2.3.
 
 This version requires the 2.1.0 `kafka-clients` or higher.
 
+==== Listener Contaienr Changes
 
+Previously, error handlers received `ListenerExectionFailedException` (with the actual listener exception as the `cause`) when the listener was invoked using a listener adapter (such as `@KafkaListener` s).
+Exceptions thrown by native `GenericMessageListener` s were passed to the error handler unchanged.
+Now a `ListenerExecutionFailedException` is always the argument (with the actual listener exception as the `cause`), which provides access to the container's `group.id` property.