spring-projectsGH-1704: More pessimistic finally

Author: Karol Dowbecki

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

@@ -63,15 +63,22 @@ default void failure(ConsumerRecords<K, V> records, Exception exception, Consume
 
 	/**
 	 * Called before consumer is polled.
+	 * <p>
+	 * It can be used to set up thread-bound resources which will be available for the
+	 * entire duration of the consumer poll operation e.g. logging with MDC.
+	 * </p>
 	 * @param consumer the consumer.
 	 * @since 2.8
 	 */
 	default void beforePoll(Consumer<K, V> consumer) {
 	}
 
 	/**
-	 * Called after listener and error handler were invoked. Last action before the
-	 * next consumer polling.
+	 * Called after listener and error handler were invoked.
+	 * <p>
+	 * It can be used to clear thread-bound resources as this is the last method called after
+	 * listener was invoked.
+	 * </p>
 	 * @param consumer the consumer.
 	 * @since 2.8
 	 */

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

@@ -1236,6 +1236,9 @@ public void run() {
 				catch (Exception e) {
 					handleConsumerException(e);
 				}
+				finally {
+					finishInvoke();
+				}
 			}
 			wrapUp(exitThrowable);
 		}
@@ -1273,6 +1276,7 @@ protected void pollAndInvoke() {
 				return;
 			}
 			this.polling.set(true);
+			beforePoll();
 			ConsumerRecords<K, V> records = doPoll();
 			if (!this.polling.compareAndSet(true, false) && records != null) {
 				/*
@@ -1291,7 +1295,6 @@ protected void pollAndInvoke() {
 			}
 
 			invokeIfHaveRecords(records);
-			finishInvoke();
 		}
 
 		private void invokeIfHaveRecords(@Nullable ConsumerRecords<K, V> records) {

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

@@ -83,15 +83,22 @@ default void failure(ConsumerRecord<K, V> record, Exception exception, Consumer<
 
 	/**
 	 * Called before consumer is polled.
+	 * <p>
+	 * It can be used to set up thread-bound resources which will be available for the
+	 * entire duration of the consumer poll operation e.g. logging with MDC.
+	 * </p>
 	 * @param consumer the consumer.
 	 * @since 2.8
 	 */
 	default void beforePoll(Consumer<K, V> consumer) {
 	}
 
 	/**
-	 * Called after listener and error handler were invoked. Last action before the
-	 * next consumer polling.
+	 * Called after listener and error handler were invoked.
+	 * <p>
+	 * It can be used to clear thread-bound resources as this is the last method called after
+	 * listener was invoked.
+	 * </p>
 	 * @param consumer the consumer.
 	 * @since 2.8
 	 */