GH-609: Pause/resume polishing; events

Resolves https://github.com/spring-projects/spring-kafka/issues/609

- Provide visibility to the container pause state
- Add pause/resume events

* Polishing - PR Comments

* Checkstyle

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

@@ -0,0 +1,54 @@
+/*
+ * 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.common.TopicPartition;
+
+/**
+ * An event published when a consumer is paused.
+ *
+ * @author Gary Russell
+ * @since 2.1.5
+ *
+ */
+@SuppressWarnings("serial")
+public class ConsumerPausedEvent extends KafkaEvent {
+
+	private final Collection<TopicPartition> partitions;
+
+	/**
+	 * Construct an instance with the provided source and partitions.
+	 * @param source the container.
+	 * @param partitions the partitions.
+	 */
+	public ConsumerPausedEvent(Object source, Collection<TopicPartition> partitions) {
+		super(source);
+		this.partitions = partitions;
+	}
+
+	public Collection<TopicPartition> getPartitions() {
+		return this.partitions;
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumerPausedEvent [partitions=" + this.partitions + "]";
+	}
+
+}

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

@@ -0,0 +1,54 @@
+/*
+ * 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.common.TopicPartition;
+
+/**
+ * An event published when a consumer is resumed.
+ *
+ * @author Gary Russell
+ * @since 2.1.5
+ *
+ */
+@SuppressWarnings("serial")
+public class ConsumerResumedEvent extends KafkaEvent {
+
+	private final Collection<TopicPartition> partitions;
+
+	/**
+	 * Construct an instance with the provided source and partitions.
+	 * @param source the container.
+	 * @param partitions the partitions.
+	 */
+	public ConsumerResumedEvent(Object source, Collection<TopicPartition> partitions) {
+		super(source);
+		this.partitions = partitions;
+	}
+
+	public Collection<TopicPartition> getPartitions() {
+		return this.partitions;
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumerResumedEvent [partitions=" + this.partitions + "]";
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -40,15 +40,44 @@ public class ListenerContainerIdleEvent extends KafkaEvent {
 
 	private final List<TopicPartition> topicPartitions;
 
+	private final boolean paused;
+
 	private transient Consumer<?, ?> consumer;
 
+	/**
+	 * Construct an instance with the provided arguments.
+	 * @param source the container.
+	 * @param idleTime the idle time.
+	 * @param id the container id.
+	 * @param topicPartitions the topics/partitions currently assigned.
+	 * @param consumer the consumer.
+	 * @deprecated in favor of
+	 * {@link #ListenerContainerIdleEvent(Object, long, String, Collection, Consumer, boolean)}
+	 */
+	@Deprecated
 	public ListenerContainerIdleEvent(Object source, long idleTime, String id,
 			Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer) {
+		this(source, idleTime, id, topicPartitions, consumer, false);
+	}
+
+	/**
+	 * Construct an instance with the provided arguments.
+	 * @param source the container.
+	 * @param idleTime the idle time.
+	 * @param id the container id.
+	 * @param topicPartitions the topics/partitions currently assigned.
+	 * @param consumer the consumer.
+	 * @param paused true if the consumer is paused.
+	 * @since 2.1.5
+	 */
+	public ListenerContainerIdleEvent(Object source, long idleTime, String id,
+			Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer, boolean paused) {
 		super(source);
 		this.idleTime = idleTime;
 		this.listenerId = id;
 		this.topicPartitions = topicPartitions == null ? null : new ArrayList<>(topicPartitions);
 		this.consumer = consumer;
+		this.paused = paused;
 	}
 
 	/**
@@ -85,11 +114,21 @@ public class ListenerContainerIdleEvent extends KafkaEvent {
 		return this.consumer;
 	}
 
+	/**
+	 * Return true if the consumer was paused at the time the idle event was published.
+	 * @return paused.
+	 * @since 2.1.5
+	 */
+	public boolean isPaused() {
+		return this.paused;
+	}
+
 	@Override
 	public String toString() {
 		return "ListenerContainerIdleEvent [idleTime="
 				+ ((float) this.idleTime / 1000) + "s, listenerId=" + this.listenerId
 				+ ", container=" + getSource()
+				+ ", paused=" + this.paused
 				+ ", topicPartitions=" + this.topicPartitions + "]";
 	}
 

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

@@ -193,6 +193,11 @@ public abstract class AbstractMessageListenerContainer<K, V>
 		return this.paused;
 	}
 
+	@Override
+	public boolean isPauseRequested() {
+		return this.paused;
+	}
+
 	public void setPhase(int phase) {
 		this.phase = phase;
 	}

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

@@ -109,6 +109,19 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 		return assigned;
 	}
 
+	@Override
+	public boolean isContainerPaused() {
+		boolean paused = isPaused();
+		if (paused) {
+			for (AbstractMessageListenerContainer<K, V> container : this.containers) {
+				if (!container.isContainerPaused()) {
+					return false;
+				}
+			}
+		}
+		return paused;
+	}
+
 	@Override
 	public Map<String, Map<MetricName, ? extends Metric>> metrics() {
 		Map<String, Map<MetricName, ? extends Metric>> metrics = new HashMap<>();

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

@@ -55,6 +55,8 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaResourceHolder;
 import org.springframework.kafka.core.ProducerFactoryUtils;
+import org.springframework.kafka.event.ConsumerPausedEvent;
+import org.springframework.kafka.event.ConsumerResumedEvent;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback;
@@ -204,6 +206,11 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		}
 	}
 
+	@Override
+	public boolean isContainerPaused() {
+		return isPaused() && this.listenerConsumer.consumerPaused;
+	}
+
 	@Override
 	public Map<String, Map<MetricName, ? extends Metric>> metrics() {
 		ListenerConsumer listenerConsumer = this.listenerConsumer;
@@ -288,21 +295,35 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		}
 	}
 
-	private void publishIdleContainerEvent(long idleTime, Consumer<?, ?> consumer) {
+	private void publishIdleContainerEvent(long idleTime, Consumer<?, ?> consumer, boolean paused) {
 		if (getApplicationEventPublisher() != null) {
 			getApplicationEventPublisher().publishEvent(new ListenerContainerIdleEvent(
-					KafkaMessageListenerContainer.this, idleTime, getBeanName(), getAssignedPartitions(), consumer));
+					this, idleTime, getBeanName(), getAssignedPartitions(), consumer, paused));
 		}
 	}
 
 	private void publishNonResponsiveConsumerEvent(long timeSinceLastPoll, Consumer<?, ?> consumer) {
 		if (getApplicationEventPublisher() != null) {
 			getApplicationEventPublisher().publishEvent(
-					new NonResponsiveConsumerEvent(KafkaMessageListenerContainer.this, timeSinceLastPoll,
+					new NonResponsiveConsumerEvent(this, timeSinceLastPoll,
 							getBeanName(), getAssignedPartitions(), consumer));
 		}
 	}
 
+	private void publishConsumerPausedEvent(Collection<TopicPartition> partitions) {
+		if (getApplicationEventPublisher() != null) {
+			getApplicationEventPublisher().publishEvent(new ConsumerPausedEvent(this,
+					Collections.unmodifiableCollection(partitions)));
+		}
+	}
+
+	private void publishConsumerResumedEvent(Collection<TopicPartition> partitions) {
+		if (getApplicationEventPublisher() != null) {
+			getApplicationEventPublisher().publishEvent(new ConsumerResumedEvent(this,
+					Collections.unmodifiableCollection(partitions)));
+		}
+	}
+
 	@Override
 	public String toString() {
 		return "KafkaMessageListenerContainer [id=" + getBeanName()
@@ -671,14 +692,17 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 						if (this.logger.isDebugEnabled()) {
 							this.logger.debug("Paused consumption from: " + this.consumer.paused());
 						}
+						publishConsumerPausedEvent(this.consumer.assignment());
 					}
 					ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());
 					if (this.consumerPaused && !isPaused()) {
 						if (this.logger.isDebugEnabled()) {
 							this.logger.debug("Resuming consumption from: " + this.consumer.paused());
 						}
-						this.consumer.resume(this.consumer.paused());
+						Set<TopicPartition> paused = this.consumer.paused();
+						this.consumer.resume(paused);
 						this.consumerPaused = false;
+						publishConsumerResumedEvent(paused);
 					}
 					if (records != null && this.logger.isDebugEnabled()) {
 						this.logger.debug("Received: " + records.count() + " records");
@@ -702,7 +726,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 							if (now > lastReceive + this.containerProperties.getIdleEventInterval()
 									&& now > lastAlertAt + this.containerProperties.getIdleEventInterval()) {
 								publishIdleContainerEvent(now - lastReceive, this.isConsumerAwareListener
-										? this.consumer : null);
+										? this.consumer : null, this.consumerPaused);
 								lastAlertAt = now;
 								if (this.genericListener instanceof ConsumerSeekAware) {
 									seekPartitions(getAssignedPartitions(), true);

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

@@ -85,4 +85,24 @@ public interface MessageListenerContainer extends SmartLifecycle {
 		throw new UnsupportedOperationException("This container doesn't support resume");
 	}
 
+	/**
+	 * Return true if {@link #pause()} has been called; the container might not have actually
+	 * paused yet.
+	 * @return true if pause has been requested.
+	 * @since 2.1.5
+	 */
+	default boolean isPauseRequested() {
+		throw new UnsupportedOperationException("This container doesn't support pause/resume");
+	}
+
+	/**
+	 * Return true if {@link #pause()} has been called; and all consumers in this container
+	 * have actually paused.
+	 * @return true if the container is paused.
+	 * @since 2.1.5
+	 */
+	default boolean isContainerPaused() {
+		throw new UnsupportedOperationException("This container doesn't support pause/resume");
+	}
+
 }

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

@@ -73,6 +73,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.event.ConsumerPausedEvent;
+import org.springframework.kafka.event.ConsumerResumedEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
 import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
@@ -1649,13 +1651,13 @@ public class KafkaMessageListenerContainerTests {
 			return null;
 		}).given(consumer).commitSync(any(Map.class));
 		given(consumer.assignment()).willReturn(records.keySet());
-		final CountDownLatch pauseLatch = new CountDownLatch(1);
+		final CountDownLatch pauseLatch = new CountDownLatch(2);
 		willAnswer(i -> {
 			pauseLatch.countDown();
 			return null;
 		}).given(consumer).pause(records.keySet());
 		given(consumer.paused()).willReturn(records.keySet());
-		final CountDownLatch resumeLatch = new CountDownLatch(1);
+		final CountDownLatch resumeLatch = new CountDownLatch(2);
 		willAnswer(i -> {
 			resumeLatch.countDown();
 			return null;
@@ -1669,6 +1671,14 @@ public class KafkaMessageListenerContainerTests {
 		containerProps.setMessageListener((MessageListener) r -> { });
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.setApplicationEventPublisher(e -> {
+			if (e instanceof ConsumerPausedEvent) {
+				pauseLatch.countDown();
+			}
+			else if (e instanceof ConsumerResumedEvent) {
+				resumeLatch.countDown();
+			}
+		});
 		container.start();
 		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		verify(consumer, times(2)).commitSync(any(Map.class));

src/reference/asciidoc/kafka.adoc

@@ -1273,7 +1273,7 @@ In addition, if the broker is unreachable (at the time of writing), the consumer
 To solve this issue, the container will publish a `NonResponsiveConsumerEvent` if a poll does not return within 3x the `pollInterval` property.
 By default, this check is performed once every 30 seconds in each container.
 You can modify the behavior by setting the `monitorInterval` and `noPollThreshold` properties in the `ContainerProperties` when configuring the listener container.
-Receiveing such an event will allow you to stop the container(s), thus waking the consumer so it can terminate.
+Receiving such an event will allow you to stop the container(s), thus waking the consumer so it can terminate.
 
 ====== Event Consumption
 
@@ -1293,7 +1293,9 @@ The events have 5 properties:
 - `topicPartitions` - the topics/partitions that the container was assigned at the time the event was generated
 - `consumer` - a reference to the kafka `Consumer` object; for example, if the consumer was previously `pause()` d, it can be `resume()` d when the event is received.
 
-The event is published on the consumer thread, so it is safe to interact with the `Consumer` object.
+Starting with _version 2.1.5_, the idle event has a boolean property `paused` which indicates whether the consumer is currently paused; see <<pause-resume>> for more information.
+
+The event is normally published on the consumer thread, so it is safe to interact with the `Consumer` object.
 
 [source, xml]
 ----
@@ -1383,6 +1385,11 @@ To safely pause/resume consumers, you should use the methods on the listener con
 `pause()` takes effect just before the next `poll()`; `resume` takes effect, just after the current `poll()` returns.
 When a container is paused, it continues to `poll()` the consumer, avoiding a rebalance if group management is being used, but will not retrieve any records; refer to the Kafka documentation for more information.
 
+Starting with _version 2.1.5_, you can call `isPauseRequested()` to see if `pause()` has been called.
+However, the consumers might not have actually paused yet; `isConsumerPaused()` will return true if all `Consumer` s have actually paused.
+
+In addition, also since _2.1.5_, `ConsumerPausedEvent` s and `ConsumerResumedEvent` s are published with the container as the `source` property and the `TopicPatition` s involved in the `partitions` s property.
+
 [[serdes]]
 ==== Serialization/Deserialization and Message Conversion