GH-99 Idle Container Events

Resolves #99

Also change `ContainerProperties` to use accessors

Also fix `stop(Runnable callback` logic for the container registry and
concurrent container.

* Polishing according PR comments and some typos fixes

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

@@ -18,6 +18,8 @@ package org.springframework.kafka.config;
 
 
 import org.springframework.beans.BeanUtils;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.adapter.DeDuplicationStrategy;
@@ -32,11 +34,12 @@ import org.springframework.kafka.support.converter.MessageConverter;
  * @param <V> the value type.
  *
  * @author Stephane Nicoll
+ * @author Gary Russell
  *
  * @see AbstractMessageListenerContainer
  */
 public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K, V>, K, V>
-		implements KafkaListenerContainerFactory<C> {
+		implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware {
 
 	private final ContainerProperties containerProperties = new ContainerProperties("propertiesFactory");
 
@@ -50,6 +53,8 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private DeDuplicationStrategy<K, V> deDuplicationStrategy;
 
+	private ApplicationEventPublisher applicationEventPublisher;
+
 	/**
 	 * Specify a {@link ConsumerFactory} to use.
 	 * @param consumerFactory The consumer factory.
@@ -96,6 +101,11 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 		this.deDuplicationStrategy = deDuplicationStrategy;
 	}
 
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 	/**
 	 * Obtain the properties template for this factory - set properties as needed
 	 * and they will be copied to a final properties instance for the endpoint.
@@ -116,6 +126,9 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 		if (this.phase != null) {
 			instance.setPhase(this.phase);
 		}
+		if (this.applicationEventPublisher != null) {
+			instance.setApplicationEventPublisher(this.applicationEventPublisher);
+		}
 		if (endpoint.getId() != null) {
 			instance.setBeanName(endpoint.getId());
 		}

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

@@ -249,7 +249,12 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 		Collection<MessageListenerContainer> listenerContainers = getListenerContainers();
 		AggregatingCallback aggregatingCallback = new AggregatingCallback(listenerContainers.size(), callback);
 		for (MessageListenerContainer listenerContainer : listenerContainers) {
-			listenerContainer.stop(aggregatingCallback);
+			if (listenerContainer.isRunning()) {
+				listenerContainer.stop(aggregatingCallback);
+			}
+			else {
+				aggregatingCallback.run();
+			}
 		}
 	}
 
@@ -297,7 +302,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 
 		@Override
 		public void run() {
-			if (this.count.decrementAndGet() == 0) {
+			if (this.count.decrementAndGet() <= 0) {
 				this.finishCallback.run();
 			}
 		}

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2015-2016 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 org.springframework.context.ApplicationEvent;
+
+
+/**
+ * Base class for events.
+ *
+ * @author Gary Russell
+ *
+ */
+@SuppressWarnings("serial")
+public abstract class KafkaEvent extends ApplicationEvent {
+
+	public KafkaEvent(Object source) {
+		super(source);
+	}
+
+}

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

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2016 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.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.kafka.common.TopicPartition;
+
+/**
+ * An event that is emitted when a container is idle if the container
+ * is configured to do so.
+ *
+ * @author Gary Russell
+ *
+ */
+@SuppressWarnings("serial")
+public class ListenerContainerIdleEvent extends KafkaEvent {
+
+	private final long idleTime;
+
+	private final String listenerId;
+
+	private final List<TopicPartition> topicPartitions;
+
+	public ListenerContainerIdleEvent(Object source, long idleTime, String id,
+			Collection<TopicPartition> topicPartitions) {
+		super(source);
+		this.idleTime = idleTime;
+		this.listenerId = id;
+		this.topicPartitions = new ArrayList<>(topicPartitions);
+	}
+
+	/**
+	 * How long the container has been idle.
+	 * @return the time in milliseconds.
+	 */
+	public long getIdleTime() {
+		return this.idleTime;
+	}
+
+	/**
+	 * The TopicPartitions the container is listening to.
+	 * @return the TopicPartition list.
+	 */
+	public Collection<TopicPartition> getTopicPartitions() {
+		return Collections.unmodifiableList(this.topicPartitions);
+	}
+
+	/**
+	 * The id of the listener (if {@code @RabbitListener}) or the container bean name.
+	 * @return the id.
+	 */
+	public String getListenerId() {
+		return this.listenerId;
+	}
+
+	@Override
+	public String toString() {
+		return "ListenerContainerIdleEvent [idleTime="
+				+ ((float) this.idleTime / 1000) + "s, listenerId=" + this.listenerId
+				+ ", container=" + getSource()
+				+ ", topicPartitions=" + this.topicPartitions + "]";
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/event/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Application Events.
+ */
+package org.springframework.kafka.event;

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

@@ -28,6 +28,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.retry.RecoveryCallback;
@@ -44,7 +46,7 @@ import org.springframework.util.Assert;
  * @author Marius Bogoevici
  */
 public abstract class AbstractMessageListenerContainer<K, V>
-		implements MessageListenerContainer, BeanNameAware, SmartLifecycle {
+		implements MessageListenerContainer, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle {
 
 	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
 
@@ -107,6 +109,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	private String beanName;
 
+	private ApplicationEventPublisher applicationEventPublisher;
+
 	private boolean autoStartup = true;
 
 	private int phase = 0;
@@ -116,8 +120,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	protected AbstractMessageListenerContainer(ContainerProperties containerProperties) {
 		Assert.notNull(containerProperties, "'containerProperties' cannot be null");
 		this.containerProperties = containerProperties;
-		if (containerProperties.consumerRebalanceListener == null) {
-			containerProperties.consumerRebalanceListener = createConsumerRebalanceListener();
+		if (containerProperties.getConsumerRebalanceListener() == null) {
+			containerProperties.setConsumerRebalanceListener(createConsumerRebalanceListener());
 		}
 	}
 
@@ -130,6 +134,15 @@ public abstract class AbstractMessageListenerContainer<K, V>
 		return this.beanName;
 	}
 
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
+	public ApplicationEventPublisher getApplicationEventPublisher() {
+		return this.applicationEventPublisher;
+	}
+
 	@Override
 	public boolean isAutoStartup() {
 		return this.autoStartup;
@@ -163,28 +176,28 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	@Override
 	public void setupMessageListener(Object messageListener) {
-		this.containerProperties.messageListener = messageListener;
+		this.containerProperties.setMessageListener(messageListener);
 	}
 
 	@Override
 	public final void start() {
 		synchronized (this.lifecycleMonitor) {
 			Assert.isTrue(
-					this.containerProperties.messageListener instanceof MessageListener
-							|| this.containerProperties.messageListener instanceof AcknowledgingMessageListener,
+					this.containerProperties.getMessageListener() instanceof MessageListener
+							|| this.containerProperties.getMessageListener() instanceof AcknowledgingMessageListener,
 					"Either a " + MessageListener.class.getName() + " or a "
 							+ AcknowledgingMessageListener.class.getName() + " must be provided");
-			if (this.containerProperties.recoveryCallback == null) {
-				this.containerProperties.recoveryCallback = new RecoveryCallback<Void>() {
+			if (this.containerProperties.getRecoveryCallback() == null) {
+				this.containerProperties.setRecoveryCallback(new RecoveryCallback<Void>() {
 
 					@Override
 					public Void recover(RetryContext context) throws Exception {
 						@SuppressWarnings("unchecked")
 						ConsumerRecord<K, V> record = (ConsumerRecord<K, V>) context.getAttribute("record");
 						Throwable lastThrowable = context.getLastThrowable();
-						if (AbstractMessageListenerContainer.this.containerProperties.errorHandler != null
+						if (AbstractMessageListenerContainer.this.containerProperties.getErrorHandler() != null
 								&& lastThrowable instanceof Exception) {
-							AbstractMessageListenerContainer.this.containerProperties.errorHandler
+							AbstractMessageListenerContainer.this.containerProperties.getErrorHandler()
 									.handle((Exception) lastThrowable, record);
 						}
 						else {
@@ -194,7 +207,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 						return null;
 					}
 
-				};
+				});
 			}
 			doStart();
 		}
@@ -212,7 +225,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 			}
 		});
 		try {
-			latch.await(this.containerProperties.shutdownTimeout, TimeUnit.MILLISECONDS);
+			latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS);
 		}
 		catch (InterruptedException e) {
 		}

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

@@ -20,6 +20,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.kafka.common.TopicPartition;
 
@@ -96,18 +97,19 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 	protected void doStart() {
 		if (!isRunning()) {
 			ContainerProperties containerProperties = getContainerProperties();
-			if (containerProperties.topicPartitions != null
-					&& this.concurrency > containerProperties.topicPartitions.length) {
+			TopicPartition[] topicPartitions = containerProperties.getTopicPartitions();
+			if (topicPartitions != null
+					&& this.concurrency > topicPartitions.length) {
 				this.logger.warn("When specific partitions are provided, the concurrency must be less than or "
 						+ "equal to the number of partitions; reduced from " + this.concurrency + " to "
-						+ containerProperties.topicPartitions.length);
-				this.concurrency = containerProperties.topicPartitions.length;
+						+ topicPartitions.length);
+				this.concurrency = topicPartitions.length;
 			}
 			setRunning(true);
 
 			for (int i = 0; i < this.concurrency; i++) {
 				KafkaMessageListenerContainer<K, V> container;
-				if (containerProperties.topicPartitions == null) {
+				if (topicPartitions == null) {
 					container = new KafkaMessageListenerContainer<>(this.consumerFactory, containerProperties);
 				}
 				else {
@@ -117,6 +119,9 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 				if (getBeanName() != null) {
 					container.setBeanName(getBeanName() + "-" + i);
 				}
+				if (getApplicationEventPublisher() != null) {
+					container.setApplicationEventPublisher(getApplicationEventPublisher());
+				}
 				container.start();
 				this.containers.add(container);
 			}
@@ -124,24 +129,23 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 	}
 
 	private TopicPartition[] partitionSubset(ContainerProperties containerProperties, int i) {
+		TopicPartition[] topicPartitions = containerProperties.getTopicPartitions();
 		if (this.concurrency == 1) {
-			return containerProperties.topicPartitions;
+			return topicPartitions;
 		}
 		else {
-			int numPartitions = containerProperties.topicPartitions.length;
+			int numPartitions = topicPartitions.length;
 			if (numPartitions == this.concurrency) {
-				return new TopicPartition[] { containerProperties.topicPartitions[i] };
+				return new TopicPartition[] { topicPartitions[i] };
 			}
 			else {
 				int perContainer = numPartitions / this.concurrency;
 				TopicPartition[] subset;
 				if (i == this.concurrency - 1) {
-					subset = Arrays.copyOfRange(containerProperties.topicPartitions, i * perContainer,
-							containerProperties.topicPartitions.length);
+					subset = Arrays.copyOfRange(topicPartitions, i * perContainer, topicPartitions.length);
 				}
 				else {
-					subset = Arrays.copyOfRange(containerProperties.topicPartitions, i * perContainer,
-							(i + 1) * perContainer);
+					subset = Arrays.copyOfRange(topicPartitions, i * perContainer, (i + 1) * perContainer);
 				}
 				return subset;
 			}
@@ -152,14 +156,37 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 	 * Under lifecycle lock.
 	 */
 	@Override
-	protected void doStop(Runnable callback) {
+	protected void doStop(final Runnable callback) {
+		final AtomicInteger count = new AtomicInteger();
 		if (isRunning()) {
 			setRunning(false);
 			for (KafkaMessageListenerContainer<K, V> container : this.containers) {
-				container.stop(callback);
+				if (container.isRunning()) {
+					count.incrementAndGet();
+				}
+			}
+			for (KafkaMessageListenerContainer<K, V> container : this.containers) {
+				if (container.isRunning()) {
+					container.stop(new Runnable() {
+
+						@Override
+						public void run() {
+							if (count.decrementAndGet() <= 0) {
+								callback.run();
+							}
+						}
+
+					});
+				}
 			}
 			this.containers.clear();
 		}
 	}
 
+	@Override
+	public String toString() {
+		return "ConcurrentMessageListenerContainer [concurrency=" + this.concurrency + ", beanName="
+				+ this.getBeanName() + ", running=" + this.isRunning() + "]";
+	}
+
 }

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

@@ -44,9 +44,11 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.retry.RetryCallback;
@@ -111,7 +113,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			this.topicPartitions = Arrays.asList(topicPartitions).toArray(new TopicPartition[topicPartitions.length]);
 		}
 		else {
-			this.topicPartitions = containerProperties.topicPartitions;
+			this.topicPartitions = containerProperties.getTopicPartitions();
 		}
 	}
 
@@ -140,7 +142,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			return;
 		}
 		setRunning(true);
-		Object messageListener = getContainerProperties().messageListener;
+		Object messageListener = getContainerProperties().getMessageListener();
 		Assert.state(messageListener != null, "A MessageListener is required");
 		if (messageListener instanceof AcknowledgingMessageListener) {
 			this.acknowledgingMessageListener = (AcknowledgingMessageListener<K, V>) messageListener;
@@ -152,19 +154,19 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			throw new IllegalStateException("messageListener must be 'MessageListener' "
 					+ "or 'AcknowledgingMessageListener', not " + messageListener.getClass().getName());
 		}
-		if (getContainerProperties().consumerTaskExecutor == null) {
+		if (getContainerProperties().getConsumerTaskExecutor() == null) {
 			SimpleAsyncTaskExecutor consumerExecutor = new SimpleAsyncTaskExecutor(
 					(getBeanName() == null ? "" : getBeanName()) + "-kafka-consumer-");
-			getContainerProperties().consumerTaskExecutor = consumerExecutor;
+			getContainerProperties().setConsumerTaskExecutor(consumerExecutor);
 		}
-		if (getContainerProperties().listenerTaskExecutor == null) {
+		if (getContainerProperties().getListenerTaskExecutor() == null) {
 			SimpleAsyncTaskExecutor listenerExecutor = new SimpleAsyncTaskExecutor(
 					(getBeanName() == null ? "" : getBeanName()) + "-kafka-listener-");
-			getContainerProperties().listenerTaskExecutor = listenerExecutor;
+			getContainerProperties().setListenerTaskExecutor(listenerExecutor);
 		}
 		this.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener,
-				getContainerProperties().recentOffset);
-		this.listenerConsumerFuture = getContainerProperties().consumerTaskExecutor
+				getContainerProperties().getRecentOffset());
+		this.listenerConsumerFuture = getContainerProperties().getConsumerTaskExecutor()
 				.submitListenable(this.listenerConsumer);
 	}
 
@@ -182,6 +184,10 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 				@Override
 				public void onSuccess(Object result) {
+					if (KafkaMessageListenerContainer.this.logger.isDebugEnabled()) {
+						KafkaMessageListenerContainer.this.logger
+								.debug(KafkaMessageListenerContainer.this + " stopped normally");
+					}
 					if (callback != null) {
 						callback.run();
 					}
@@ -192,14 +198,22 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		}
 	}
 
+
+	@Override
+	public String toString() {
+		return "KafkaMessageListenerContainer [id=" + getBeanName() + ", topicPartitions=" + getAssignedPartitions()
+				+ "]";
+	}
+
+
 	private final class ListenerConsumer implements SchedulingAwareRunnable {
 
 		private final Log logger = LogFactory.getLog(ListenerConsumer.class);
 
 		private final ContainerProperties containerProperties = getContainerProperties();
 
-		private final OffsetCommitCallback commitCallback = this.containerProperties.commitCallback != null
-				? this.containerProperties.commitCallback
+		private final OffsetCommitCallback commitCallback = this.containerProperties.getCommitCallback() != null
+				? this.containerProperties.getCommitCallback()
 				: new LoggingCommitCallback();
 
 		private final Consumer<K, V> consumer;
@@ -216,20 +230,23 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 		private final boolean autoCommit = KafkaMessageListenerContainer.this.consumerFactory.isAutoCommit();
 
-		private final boolean isManualAck = this.containerProperties.ackMode.equals(AckMode.MANUAL);
+		private final boolean isManualAck = this.containerProperties.getAckMode().equals(AckMode.MANUAL);
 
-		private final boolean isManualImmediateAck = this.containerProperties.ackMode.equals(AckMode.MANUAL_IMMEDIATE)
-				|| this.containerProperties.ackMode.equals(AckMode.MANUAL_IMMEDIATE_SYNC);
+		private final boolean isManualImmediateAck =
+				this.containerProperties.getAckMode().equals(AckMode.MANUAL_IMMEDIATE)
+				|| this.containerProperties.getAckMode().equals(AckMode.MANUAL_IMMEDIATE_SYNC);
 
 		private final boolean isAnyManualAck = this.isManualAck || this.isManualImmediateAck;
 
-		private final boolean isRecordAck = this.containerProperties.ackMode.equals(AckMode.RECORD);
+		private final boolean isRecordAck = this.containerProperties.getAckMode().equals(AckMode.RECORD);
 
-		private final BlockingQueue<ConsumerRecords<K, V>> recordsToProcess = new LinkedBlockingQueue<>(
-				this.containerProperties.queueDepth);
+		private final BlockingQueue<ConsumerRecords<K, V>> recordsToProcess =
+				new LinkedBlockingQueue<>(this.containerProperties.getQueueDepth());
 
 		private final BlockingQueue<ConsumerRecord<K, V>> acks = new LinkedBlockingQueue<>();
 
+		private final ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
+
 		private volatile Collection<TopicPartition> definedPartitions;
 
 		private volatile Collection<TopicPartition> assignedPartitions;
@@ -252,7 +269,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		private ListenerConsumer(MessageListener<K, V> listener, AcknowledgingMessageListener<K, V> ackListener,
 				long recentOffset) {
 			Assert.state(!this.isAnyManualAck || !this.autoCommit,
-					"Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.ackMode);
+				"Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode());
 			Consumer<K, V> consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer();
 
 			ConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() {
@@ -284,7 +301,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 											+ "autocommit mode, so transition will be handled by the consumer");
 						}
 					}
-					getContainerProperties().consumerRebalanceListener.onPartitionsRevoked(partitions);
+					getContainerProperties().getConsumerRebalanceListener().onPartitionsRevoked(partitions);
 				}
 
 				@Override
@@ -299,17 +316,17 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 							&& !CollectionUtils.isEmpty(partitions)) {
 						startInvoker();
 					}
-					getContainerProperties().consumerRebalanceListener.onPartitionsAssigned(partitions);
+					getContainerProperties().getConsumerRebalanceListener().onPartitionsAssigned(partitions);
 				}
 
 			};
 
 			if (KafkaMessageListenerContainer.this.topicPartitions == null) {
-				if (this.containerProperties.topicPattern != null) {
-					consumer.subscribe(this.containerProperties.topicPattern, rebalanceListener);
+				if (this.containerProperties.getTopicPattern() != null) {
+					consumer.subscribe(this.containerProperties.getTopicPattern(), rebalanceListener);
 				}
 				else {
-					consumer.subscribe(Arrays.asList(this.containerProperties.topics), rebalanceListener);
+					consumer.subscribe(Arrays.asList(this.containerProperties.getTopics()), rebalanceListener);
 				}
 			}
 			else {
@@ -326,7 +343,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 		private void startInvoker() {
 			ListenerConsumer.this.invoker = new ListenerInvoker();
-			ListenerConsumer.this.listenerInvokerFuture = this.containerProperties.listenerTaskExecutor
+			ListenerConsumer.this.listenerInvokerFuture = this.containerProperties.getListenerTaskExecutor()
 					.submit(ListenerConsumer.this.invoker);
 		}
 
@@ -349,16 +366,21 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 				}
 			}
 			ConsumerRecords<K, V> unsent = null;
+			long lastReceive = System.currentTimeMillis();
+			long lastAlertAt = lastReceive;
 			while (isRunning()) {
 				try {
 					if (this.logger.isTraceEnabled()) {
 						this.logger.trace("Polling (paused=" + this.paused + ")...");
 					}
-					ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.pollTimeout);
+					ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());
 					if (this.logger.isDebugEnabled()) {
 						this.logger.debug("Received: " + records.count() + " records");
 					}
 					if (records != null && records.count() > 0) {
+						if (this.containerProperties.getIdleEventInterval() != null) {
+							lastReceive = System.currentTimeMillis();
+						}
 						handleManualAcks();
 						// if the container is set to auto-commit, then execute in the
 						// same thread
@@ -379,6 +401,16 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 							}
 						}
 					}
+					else {
+						if (this.containerProperties.getIdleEventInterval() != null) {
+							long now = System.currentTimeMillis();
+							if (now > lastReceive + this.containerProperties.getIdleEventInterval()
+									&& now > lastAlertAt + this.containerProperties.getIdleEventInterval()) {
+								publishIdleContainerEvent(now - lastReceive);
+								lastAlertAt = now;
+							}
+						}
+					}
 					unsent = checkPause(unsent);
 					if (!this.paused && !this.autoCommit) {
 						processCommits();
@@ -388,8 +420,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					unsent = checkPause(unsent);
 				}
 				catch (Exception e) {
-					if (this.containerProperties.errorHandler != null) {
-						this.containerProperties.errorHandler.handle(e, null);
+					if (this.containerProperties.getErrorHandler() != null) {
+						this.containerProperties.getErrorHandler().handle(e, null);
 					}
 					else {
 						this.logger.error("Container exception", e);
@@ -411,10 +443,17 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			}
 		}
 
+		private void publishIdleContainerEvent(long idleTime) {
+			if (this.applicationEventPublisher != null) {
+				this.applicationEventPublisher.publishEvent(new ListenerContainerIdleEvent(
+						KafkaMessageListenerContainer.this, idleTime, getBeanName(), getAssignedPartitions()));
+			}
+		}
+
 		private void stopInvokerAndCommitManualAcks() {
 			this.invoker.stop();
 			try {
-				this.listenerInvokerFuture.get(this.containerProperties.shutdownTimeout, TimeUnit.MILLISECONDS);
+				this.listenerInvokerFuture.get(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS);
 			}
 			catch (InterruptedException e) {
 				Thread.currentThread().interrupt();
@@ -439,7 +478,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		}
 
 		private ConsumerRecords<K, V> checkPause(ConsumerRecords<K, V> unsent) {
-			if (this.paused && this.recordsToProcess.size() < this.containerProperties.queueDepth) {
+			if (this.paused && this.recordsToProcess.size() < this.containerProperties.getQueueDepth()) {
 				// Listener has caught up.
 				this.consumer.resume(
 						this.assignedPartitions.toArray(new TopicPartition[this.assignedPartitions.size()]));
@@ -459,8 +498,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		}
 
 		private boolean sendToListener(final ConsumerRecords<K, V> records) throws InterruptedException {
-			if (this.containerProperties.pauseEnabled && CollectionUtils.isEmpty(this.definedPartitions)) {
-				return !this.recordsToProcess.offer(records, this.containerProperties.pauseAfter,
+			if (this.containerProperties.isPauseEnabled() && CollectionUtils.isEmpty(this.definedPartitions)) {
+				return !this.recordsToProcess.offer(records, this.containerProperties.getPauseAfter(),
 						TimeUnit.MILLISECONDS);
 			}
 			else {
@@ -503,7 +542,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			if (ListenerConsumer.this.logger.isDebugEnabled()) {
 				ListenerConsumer.this.logger.debug("Committing: " + commits);
 			}
-			if (ListenerConsumer.this.containerProperties.ackMode.equals(AckMode.MANUAL_IMMEDIATE)) {
+			if (ListenerConsumer.this.containerProperties.getAckMode().equals(AckMode.MANUAL_IMMEDIATE)) {
 				ListenerConsumer.this.consumer.commitAsync(commits,
 						ListenerConsumer.this.commitCallback);
 			}
@@ -521,8 +560,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 				}
 				try {
 					if (this.acknowledgingMessageListener != null) {
-						if (this.containerProperties.retryTemplate != null) {
-							this.containerProperties.retryTemplate.execute(new RetryCallback<Void, KafkaException>() {
+						if (this.containerProperties.getRetryTemplate() != null) {
+							this.containerProperties.getRetryTemplate().execute(
+									new RetryCallback<Void, KafkaException>() {
 
 								@Override
 								public Void doWithRetry(RetryContext context) throws KafkaException {
@@ -532,15 +572,16 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 									return null;
 								}
 
-							}, this.containerProperties.recoveryCallback);
+							}, this.containerProperties.getRecoveryCallback());
 						}
 						else {
 							this.acknowledgingMessageListener.onMessage(record, new ConsumerAcknowledgment(record));
 						}
 					}
 					else {
-						if (this.containerProperties.retryTemplate != null) {
-							this.containerProperties.retryTemplate.execute(new RetryCallback<Void, KafkaException>() {
+						if (this.containerProperties.getRetryTemplate() != null) {
+							this.containerProperties.getRetryTemplate().execute(
+									new RetryCallback<Void, KafkaException>() {
 
 								@Override
 								public Void doWithRetry(RetryContext context) throws KafkaException {
@@ -549,7 +590,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 									return null;
 								}
 
-							}, this.containerProperties.recoveryCallback);
+							}, this.containerProperties.getRecoveryCallback());
 						}
 						else {
 							this.listener.onMessage(record);
@@ -561,8 +602,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					}
 				}
 				catch (Exception e) {
-					if (this.containerProperties.errorHandler != null) {
-						this.containerProperties.errorHandler.handle(e, record);
+					if (this.containerProperties.getErrorHandler() != null) {
+						this.containerProperties.getErrorHandler().handle(e, record);
 					}
 					else {
 						this.logger.error("Listener threw an exception and no error handler for " + record, e);
@@ -574,19 +615,19 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		private void processCommits() {
 			this.count += this.acks.size();
 			long now;
-			AckMode ackMode = this.containerProperties.ackMode;
+			AckMode ackMode = this.containerProperties.getAckMode();
 			if (!this.isManualImmediateAck) {
 				if (!this.isManualAck) {
 					updatePendingOffsets();
 				}
-				boolean countExceeded = this.count >= this.containerProperties.ackCount;
+				boolean countExceeded = this.count >= this.containerProperties.getAckCount();
 				if (ackMode.equals(AckMode.BATCH) || ackMode.equals(AckMode.COUNT) && countExceeded) {
 					commitIfNecessary();
 					this.count = 0;
 				}
 				else {
 					now = System.currentTimeMillis();
-					boolean elapsed = now - this.last > this.containerProperties.ackTime;
+					boolean elapsed = now - this.last > this.containerProperties.getAckTime();
 					if (ackMode.equals(AckMode.TIME) && elapsed) {
 						commitIfNecessary();
 						this.last = now;
@@ -669,7 +710,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			}
 			if (!commits.isEmpty()) {
 				try {
-					if (this.containerProperties.syncCommits) {
+					if (this.containerProperties.isSyncCommits()) {
 						this.consumer.commitSync(commits);
 					}
 					else {

spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java

@@ -51,17 +51,17 @@ public class ContainerProperties {
 	/**
 	 * Topic names.
 	 */
-	public final String[] topics;
+	private final String[] topics;
 
 	/**
 	 * Topic pattern.
 	 */
-	public final Pattern topicPattern;
+	private final Pattern topicPattern;
 
 	/**
 	 * Topics/partitions.
 	 */
-	public final TopicPartition[] topicPartitions;
+	private final TopicPartition[] topicPartitions;
 
 	/**
 	 * The ack mode to use when auto ack (in the configuration properties) is false.
@@ -76,102 +76,102 @@ public class ContainerProperties {
 	 * {@link AcknowledgingMessageListener}.
 	 * </ul>
 	 */
-	public AbstractMessageListenerContainer.AckMode ackMode = AckMode.BATCH;
+	private AbstractMessageListenerContainer.AckMode ackMode = AckMode.BATCH;
 
 	/**
 	 * The number of outstanding record count after which offsets should be
 	 * committed when {@link AckMode#COUNT} or {@link AckMode#COUNT_TIME} is being
 	 * used.
 	 */
-	public int ackCount;
+	private int ackCount;
 
 	/**
 	 * The time (ms) after which outstanding offsets should be committed when
 	 * {@link AckMode#TIME} or {@link AckMode#COUNT_TIME} is being used. Should be
 	 * larger than
 	 */
-	public long ackTime;
+	private long ackTime;
 
 	/**
 	 * The message listener; must be a {@link MessageListener} or
 	 * {@link AcknowledgingMessageListener}.
 	 */
-	public Object messageListener;
+	private Object messageListener;
 
 	/**
 	 * The max time to block in the consumer waiting for records.
 	 */
-	public volatile long pollTimeout = 1000;
+	private volatile long pollTimeout = 1000;
 
 	/**
 	 * The executor for threads that poll the consumer.
 	 */
-	public AsyncListenableTaskExecutor consumerTaskExecutor;
+	private AsyncListenableTaskExecutor consumerTaskExecutor;
 
 	/**
 	 * The executor for threads that invoke the listener.
 	 */
-	public AsyncListenableTaskExecutor listenerTaskExecutor;
+	private AsyncListenableTaskExecutor listenerTaskExecutor;
 
 	/**
 	 * The error handler to call when the listener throws an exception.
 	 */
-	public ErrorHandler errorHandler = new LoggingErrorHandler();
+	private ErrorHandler errorHandler = new LoggingErrorHandler();
 
 	/**
 	 * When using Kafka group management and {@link #setPauseEnabled(boolean)} is
 	 * true, the delay after which the consumer should be paused. Default 10000.
 	 */
-	public long pauseAfter = DEFAULT_PAUSE_AFTER;
+	private long pauseAfter = DEFAULT_PAUSE_AFTER;
 
 	/**
 	 * When true, avoids rebalancing when this consumer is slow or throws a
 	 * qualifying exception - pauses the consumer. Default: true.
 	 * @see #pauseAfter
 	 */
-	public boolean pauseEnabled = true;
+	private boolean pauseEnabled = true;
 
 	/**
 	 * A retry template to retry deliveries.
 	 */
-	public RetryTemplate retryTemplate;
+	private RetryTemplate retryTemplate;
 
 	/**
 	 * A recovery callback to be invoked when retries are exhausted. By default
 	 * the error handler is invoked.
 	 */
-	public RecoveryCallback<Void> recoveryCallback;
+	private RecoveryCallback<Void> recoveryCallback;
 
 	/**
 	 * Set the queue depth for handoffs from the consumer thread to the listener
 	 * thread. Default 1 (up to 2 in process).
 	 */
-	public int queueDepth = DEFAULT_QUEUE_DEPTH;
+	private int queueDepth = DEFAULT_QUEUE_DEPTH;
 
 	/**
 	 * The timeout for shutting down the container. This is the maximum amount of
 	 * time that the invocation to {@code #stop(Runnable)} will block for, before
 	 * returning.
 	 */
-	public long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;
+	private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;
 
 	/**
 	 * The offset to this number of records back from the latest when starting.
 	 * Overrides any consumer properties (earliest, latest). Only applies when
 	 * explicit topic/partition assignment is provided.
 	 */
-	public long recentOffset;
+	private long recentOffset;
 
 	/**
 	 * A user defined {@link ConsumerRebalanceListener} implementation.
 	 */
-	public ConsumerRebalanceListener consumerRebalanceListener;
+	private ConsumerRebalanceListener consumerRebalanceListener;
 
 	/**
 	 * The commit callback; by default a simple logging callback is used to log
 	 * success at DEBUG level and failures at ERROR level.
 	 */
-	public OffsetCommitCallback commitCallback;
+	private OffsetCommitCallback commitCallback;
 
 	/**
 	 * Whether or not to call consumer.commitSync() or commitAsync() when the
@@ -179,7 +179,9 @@ public class ContainerProperties {
 	 * https://github.com/spring-projects/spring-kafka/issues/62 At the time of
 	 * writing, async commits are not entirely reliable.
 	 */
-	public boolean syncCommits = true;
+	private boolean syncCommits = true;
+
+	private Long idleEventInterval;
 
 	public ContainerProperties(String... topics) {
 		Assert.notEmpty(topics, "An array of topicPartitions must be provided");
@@ -374,10 +376,14 @@ public class ContainerProperties {
 		this.syncCommits = syncCommits;
 	}
 
-	/*
-	 * Although we generally use field access, we need the getters so we can copy
-	 * the properties from the factory when creating an annotated listener.
+	/**
+	 * Set the idle event interval; when set, an event is emitted if a poll returns
+	 * no records and this interval has elapsed since a record was returned.
+	 * @param idleEventInterval the interval.
 	 */
+	public void setIdleEventInterval(Long idleEventInterval) {
+		this.idleEventInterval = idleEventInterval;
+	}
 
 	public String[] getTopics() {
 		return this.topics;
@@ -463,4 +469,8 @@ public class ContainerProperties {
 		return this.syncCommits;
 	}
 
+	public Long getIdleEventInterval() {
+		return this.idleEventInterval;
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -36,6 +36,7 @@ import org.mockito.Mockito;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.EventListener;
 import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
@@ -45,6 +46,7 @@ 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.ListenerContainerIdleEvent;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.MessageListenerContainer;
@@ -124,6 +126,8 @@ public class EnableKafkaIntegrationTests {
 		assertThat(this.listener.latch4.await(20, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.listener.record.value()).isEqualTo("foo");
 		assertThat(this.listener.ack).isNotNull();
+		assertThat(this.listener.eventLatch.await(20, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.event.getListenerId().startsWith("qux-"));
 
 		template.send("annotated5", 0, 0, "foo");
 		template.send("annotated5", 1, 0, "bar");
@@ -236,6 +240,7 @@ public class EnableKafkaIntegrationTests {
 			factory.setConsumerFactory(manualConsumerFactory());
 			ContainerProperties props = factory.getContainerProperties();
 			props.setAckMode(AckMode.MANUAL_IMMEDIATE);
+			props.setIdleEventInterval(100L);
 			return factory;
 		}
 
@@ -354,6 +359,8 @@ public class EnableKafkaIntegrationTests {
 
 		private final CountDownLatch latch7 = new CountDownLatch(1);
 
+		private final CountDownLatch eventLatch = new CountDownLatch(1);
+
 		private volatile Integer partition;
 
 		private volatile ConsumerRecord<?, ?> record;
@@ -366,6 +373,8 @@ public class EnableKafkaIntegrationTests {
 
 		private Foo foo;
 
+		private volatile ListenerContainerIdleEvent event;
+
 		@KafkaListener(id = "manualStart", topics = "manualStart",
 				containerFactory = "kafkaAutoStartFalseListenerContainerFactory")
 		public void manualStart(String foo) {
@@ -401,6 +410,12 @@ public class EnableKafkaIntegrationTests {
 			this.latch4.countDown();
 		}
 
+		@EventListener(condition = "event.listenerId.startsWith('qux')")
+		public void eventHandler(ListenerContainerIdleEvent event) {
+			this.event = event;
+			eventLatch.countDown();
+		}
+
 		@KafkaListener(id = "fiz", topicPartitions = {
 				@TopicPartition(topic = "annotated5", partitions = { "#{'${foo:0,1}'.split(',')}" }),
 				@TopicPartition(topic = "annotated6", partitions = { "0", "1" })
@@ -442,7 +457,7 @@ public class EnableKafkaIntegrationTests {
 			latch1.countDown();
 		}
 
-		@KafkaListener(topics = "annotated9")
+		@KafkaListener(id = "ifctx", topics = "annotated9")
 		@Transactional
 		public void listenTx(String foo) {
 			latch2.countDown();
@@ -457,7 +472,7 @@ public class EnableKafkaIntegrationTests {
 		}
 	}
 
-	@KafkaListener(topics = "annotated8")
+	@KafkaListener(id = "multi", topics = "annotated8")
 	static class MultiListenerBean {
 
 		private final CountDownLatch latch1 = new CountDownLatch(1);

src/reference/asciidoc/kafka.adoc

@@ -149,7 +149,7 @@ public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                                                         ContainerProperties containerProperties)
 
 public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
-                        ContainerProperties containerProperties, TopicPartition... topicPartitions) {
+                        ContainerProperties containerProperties, TopicPartition... topicPartitions)
 
 ----
 
@@ -228,13 +228,13 @@ NOTE: `MANUAL`, `MANUAL_IMMEDIATE`, and `MANUAL_IMMEDIATE_SYNC` require the list
 ----
 public interface AcknowledgingMessageListener<K, V> {
 
-	void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment);
+    void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment);
 
 }
 
 public interface Acknowledgment {
 
-	void acknowledge();
+    void acknowledge();
 
 }
 ----
@@ -409,3 +409,87 @@ public void jsonListener(Foo foo) {
 ...
 }
 ----
+
+[[idle-containers]]
+===== Detecting Idle Asynchronous Consumers
+
+While efficient, one problem with asynchronous consumers is detecting when they are idle - users might want to take
+some action if no messages arrive for some period of time.
+
+You can configure the listener container to publish a `ListenerContainerIdleEvent` when some time passes with no message delivery.
+While the container is idle, an event will be published every `idleEventInterval` milliseconds.
+
+To configure this feature, set the `idleEventInterval` on the container:
+
+[source, java]
+----
+@Bean
+public KafKaMessageListenerContainer(ConnectionFactory connectionFactory) {
+    ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
+    ...
+    containerProps.setIdleEventInterval(60000L);
+    ...
+    KafKaMessageListenerContainer<String, String> container = new KafKaMessageListenerContainer<>(...);
+    return container;
+}
+----
+
+Or, for a `@KafkaListener`...
+
+[source, java]
+----
+@Bean
+public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
+    ConcurrentKafkaListenerContainerFactory<String, String> factory =
+                new ConcurrentKafkaListenerContainerFactory<>();
+    ...
+    factory.getContainerProperties().setIdleEventInterval(60000L);
+    ...
+    return factory;
+}
+----
+
+In each of these cases, an event will be published once per minute while the container is idle.
+
+====== Event Consumption
+
+You can capture these events by implementing `ApplicationListener` - either a general listener, or one narrowed to only receive this specific event.
+You can also use `@EventListener`, introduced in Spring Framework 4.2.
+
+The following example combines the `@KafkaListener` and `@EventListener` into a single class.
+It's important to understand that the application listener will get events for all containers so you may need to
+check the listener id if you want to take specific action based on which container is idle.
+You can also use the `@EventListener` `condition` for this purpose.
+
+The events have 4 properties:
+
+- `source` - the listener container instance
+- `id` - the listener id (or container bean name)
+- `idleTime` - the time the container had been idle when the event was published
+- `topicPartitions` - the topics/partitions that the container was assigned at the time the event was generated
+
+[source, xml]
+----
+public class Listener {
+
+    @KafkaListener(id = "qux", topics = "annotated")
+    public void listen4(@Payload String foo, Acknowledgment ack) {
+        ...
+    }
+
+    @EventListener(condition = "event.listenerId.startsWith('qux-')")
+    public void eventHandler(ListenerContainerIdleEvent event) {
+        this.event = event;
+        eventLatch.countDown();
+    }
+
+}
+----
+
+IMPORTANT: Event listeners will see events for all containers; so, in the example above, we narrow the events received based on the listener ID.
+Since containers created for the `@KafkaListener` support concurrency, the actual containers are named `id-n` where the `n` is a unique value for each instance to support the concurrency.
+Hence we use `startsWith` in the condition.
+
+CAUTION: If you wish to use the idle event to stop the lister container, you should not call `container.stop()` on the thread that calls the listener - it will cause delays and unnecessary log messages.
+Instead, you should hand off the event to a different thread that can then stop the container.
+Also, you should not `stop()` the container instance in the event if it is a child container, you should stop the concurrent container instead.