GH-867: Add ConsumerStoppingEvent

Resolves #867

Author: garyrussell

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

@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+import java.util.Collection;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+
+/**
+ * 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 ConsumerStoppingEvent extends KafkaEvent {
+
+	private final Consumer<?, ?> consumer;
+
+	private final Collection<TopicPartition> partitions;
+
+	/**
+	 * Construct an instance with the provided source, consumer and partitions.
+	 * @param source the container.
+	 * @param consumer the consumer.
+	 * @param partitions the partitions.
+	 */
+	public ConsumerStoppingEvent(Object source, Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
+		super(source);
+		this.consumer = consumer;
+		this.partitions = partitions;
+	}
+
+	public Consumer<?, ?> getConsumer() {
+		return this.consumer;
+	}
+
+	public Collection<TopicPartition> getPartitions() {
+		return this.partitions;
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumerStoppingEvent [consumer=" + this.consumer + ", partitions=" + this.partitions + "]";
+	}
+
+}

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

@@ -66,6 +66,7 @@
 import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.event.ConsumerResumedEvent;
 import org.springframework.kafka.event.ConsumerStoppedEvent;
+import org.springframework.kafka.event.ConsumerStoppingEvent;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback;
@@ -331,6 +332,17 @@ private void publishConsumerResumedEvent(Collection<TopicPartition> partitions)
 		}
 	}
 
+	private void publishConsumerStoppingEvent(Consumer<?, ?> consumer) {
+		try {
+			if (getApplicationEventPublisher() != null) {
+				getApplicationEventPublisher().publishEvent(
+						new ConsumerStoppingEvent(this, consumer, getAssignedPartitions()));
+			}
+		}
+		catch (Exception e) {
+			this.logger.error("Failed to publish consumer stopping event", e);
+		}
+	}
 	private void publishConsumerStoppedEvent() {
 		if (getApplicationEventPublisher() != null) {
 			getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this));
@@ -721,6 +733,7 @@ public void run() {
 
 		public void wrapUp() {
 			ProducerFactoryUtils.clearConsumerGroupId();
+			publishConsumerStoppingEvent(this.consumer);
 			if (!this.fatalError) {
 				if (this.kafkaTxManager == null) {
 					commitPendingAcks();

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

@@ -46,6 +46,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -78,6 +79,7 @@
 import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.event.ConsumerResumedEvent;
 import org.springframework.kafka.event.ConsumerStoppedEvent;
+import org.springframework.kafka.event.ConsumerStoppingEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
@@ -173,6 +175,16 @@ public void testDelegateType() throws Exception {
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.setBeanName("delegate");
+		AtomicReference<List<TopicPartitionInitialOffset>> offsets = new AtomicReference<>();
+		container.setApplicationEventPublisher(e -> {
+			if (e instanceof ConsumerStoppingEvent) {
+				ConsumerStoppingEvent event = (ConsumerStoppingEvent) e;
+				offsets.set(event.getPartitions().stream()
+						.map(p -> new TopicPartitionInitialOffset(p.topic(), p.partition(),
+								event.getConsumer().position(p, Duration.ofMillis(10_000))))
+						.collect(Collectors.toList()));
+			}
+		});
 		container.start();
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
@@ -186,6 +198,16 @@ public void testDelegateType() throws Exception {
 		// Stack traces are environment dependent - verified in eclipse
 		//		assertThat(trace.get()[1].getMethodName()).contains("invokeRecordListener");
 		container.stop();
+		List<TopicPartitionInitialOffset> list = offsets.get();
+		assertThat(list).isNotNull();
+		list.forEach(tpio -> {
+				if (tpio.partition() == 0) {
+					assertThat(tpio.initialOffset()).isEqualTo(1);
+				}
+				else {
+					assertThat(tpio.initialOffset()).isEqualTo(0);
+				}
+		});
 		final CountDownLatch latch2 = new CountDownLatch(1);
 		FilteringMessageListenerAdapter<Integer, String> filtering = new FilteringMessageListenerAdapter<>(m -> {
 			trace.set(new RuntimeException().getStackTrace());
@@ -206,7 +228,6 @@ public void testDelegateType() throws Exception {
 		//		assertThat(trace.get()[5].getMethodName()).contains("onMessage"); // bridge
 		//		assertThat(trace.get()[6].getMethodName()).contains("invokeRecordListener");
 		container.stop();
-
 		final CountDownLatch latch3 = new CountDownLatch(1);
 		filtering = new FilteringMessageListenerAdapter<>(
 				(AcknowledgingConsumerAwareMessageListener<Integer, String>) (d, a, c) -> {
@@ -2210,10 +2231,12 @@ public Bar(String baz) {
 			this.baz = baz;
 		}
 
+		@SuppressWarnings("unused")
 		private String getBaz() {
 			return this.baz;
 		}
 
+		@SuppressWarnings("unused")
 		private void setBaz(String baz) {
 			this.baz = baz;
 		}
@@ -2237,10 +2260,12 @@ public Bar1(String baz) {
 			this.baz = baz;
 		}
 
+		@SuppressWarnings("unused")
 		private String getBaz() {
 			return this.baz;
 		}
 
+		@SuppressWarnings("unused")
 		private void setBaz(String baz) {
 			this.baz = baz;
 		}

src/reference/asciidoc/kafka.adoc

@@ -1777,6 +1777,23 @@ ConsumerResumedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
 bar
 ----
 
+[[events]]
+==== Events
+
+The following events are published by listener containers and their consumers:
+
+* `ContainerIdleEvent` - when no messages have been received in `idleInterval` (if configured)
+* `NonResponsiveConsumerEvent` - when the consumer appears to be blocked in the `poll` method
+* `ConsumerPausedEvent` - issued by each consumer when the container is paused
+* `ConsumerResumedEvent` - issued by each consumer when the container is resumed
+* `ConsumerStoppingEvent` - issued by each consumer just before stopping
+* `ConsumerStoppedEvent` - issued after the consumer is closed; see <<thread-safety>>
+* `ContainerStoppedEvent` - when all consumers have terminated
+
+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, you must not invoke any `Consumer` methods when the event contains a reference to the consumer.
+
+
 [[serdes]]
 ==== Serialization/Deserialization and Message Conversion
 

src/reference/asciidoc/whats-new.adoc

@@ -40,6 +40,9 @@ They can be configured to publish failed records to a dead-letter topic.
 
 See <<after-rollback>>, <<seek-to-current>> and <<dead-letters>> for more information.
 
+The `ConsumerStoppingEvent` has been added.
+See <<events>> for more information.
+
 ==== @KafkaListener Changes
 
 You can now override the `concurrency` and `autoStartup` properties of the listener container factory by setting properties on the annotation.