GH-762: ConsumerStoppedEvent

Resolves #762

Add an event when a consumer thread exits.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStoppedEvent.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.event;
+
+/**
+ * An event published when a consumer is stopped. While it is best practice to use
+ * stateless listeners, you can consume this event to clean up any thread-based resources
+ * (remove ThreadLocals, destroy thread-scoped beans etc), as long as the context event
+ * multicaster is not modified to use an async task executor.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@SuppressWarnings("serial")
+public class ConsumerStoppedEvent extends KafkaEvent {
+
+	/**
+	 * Construct an instance with the provided source and partitions.
+	 * @param source the container.
+	 */
+	public ConsumerStoppedEvent(Object source) {
+		super(source);
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumerStoppedEvent [source=" + getSource() + "]";
+	}
+
+}

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

@@ -58,6 +58,7 @@
 import org.springframework.kafka.core.ProducerFactoryUtils;
 import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.event.ConsumerResumedEvent;
+import org.springframework.kafka.event.ConsumerStoppedEvent;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback;
@@ -326,6 +327,12 @@ private void publishConsumerResumedEvent(Collection<TopicPartition> partitions)
 		}
 	}
 
+	private void publishConsumerStoppedEvent() {
+		if (getApplicationEventPublisher() != null) {
+			getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this));
+		}
+	}
+
 	@Override
 	public String toString() {
 		return "KafkaMessageListenerContainer [id=" + getBeanName()
@@ -783,6 +790,7 @@ public void run() {
 			}
 			this.consumer.close();
 			this.logger.info("Consumer stopped");
+			publishConsumerStoppedEvent();
 		}
 
 		/**

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

@@ -76,6 +76,7 @@
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.event.ConsumerResumedEvent;
+import org.springframework.kafka.event.ConsumerStoppedEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
@@ -1782,13 +1783,17 @@ public void testPauseResume() throws Exception {
 		containerProps.setMessageListener((MessageListener) r -> { });
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
+		CountDownLatch stopLatch = new CountDownLatch(1);
 		container.setApplicationEventPublisher(e -> {
 			if (e instanceof ConsumerPausedEvent) {
 				pauseLatch.countDown();
 			}
 			else if (e instanceof ConsumerResumedEvent) {
 				resumeLatch.countDown();
 			}
+			else if (e instanceof ConsumerStoppedEvent) {
+				stopLatch.countDown();
+			}
 		});
 		container.start();
 		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -1798,6 +1803,7 @@ else if (e instanceof ConsumerResumedEvent) {
 		container.resume();
 		assertThat(resumeLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		container.stop();
+		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })

src/reference/asciidoc/kafka.adoc

@@ -1475,6 +1475,24 @@ public ConcurrentMessageListenerContainer<String, String>(
 IMPORTANT: Containers created this way are not added to the endpoint registry.
 They should be created as `@Bean` s so that they will be registered with the application context.
 
+[[thread-safety]]
+===== Thread Safety
+
+When using a concurrent message listener container, a single listener instance is invoked on all consumer threads.
+Listeners, therefore, need to be thread-safe; and it is preferable to use stateless listeners.
+If it is not possible to make your listener thread-safe, or adding synchronization would significantly reduce the benefit of adding concurreny, there are several techniques you can use.
+
+. Use `n` containers with `concurrency=1` with a prototype scoped `MessageListener` bean so each container gets its own instance (this is not possible when using `@KafkaListener`).
+. Keep the state in `ThreadLocal<?>` s.
+. Have the singleton listener delegate to a bean that is declared in `SimpleThreadScope` or similar.
+
+To facilitate cleaning up thread state (for 2 and 3), starting with version 2.2, the listener container will publish `ConsumerStoppedEvent` s when each thread exits.
+Consume these events with an `ApplicationListener` or `@EventListener` method to remove `ThreadLocal<?>` s, or `remove()` thread-scoped beans from the scope.
+Note that `SimpleThreadScope` does not destroy beans that have a destruction interface (e.g. `DisposableBean`) so you should `destroy()` the instance yourself.
+
+IMPORTANT: By default, the application context's event multicaster invokes event listeners on the calling thread.
+If you change the multicaster to use an async executor, thread cleanup will not be effective.
+
 [[pause-resume]]
 ==== Pausing/Resuming Listener Containers
 

src/reference/asciidoc/whats-new.adoc

@@ -27,6 +27,10 @@ A new container property `missingTopicsFatal` has been added.
 
 See <<kafka-container>> for more information.
 
+A `ConsumerStoppedEvent` is now emitted when a consumer terminates.
+
+See <<thread-safety>> for more information.
+
 Batch listeners can optionally receive the complete `ConsumerRecords<?, ?>` object instead of a `List<ConsumerRecord<?, ?>`.
 
 See <<batch-listeners>> for more information.