GH-1252: Add more consumer lifecycle events

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

- also log an error when a consumer fails to start

* Fix publish.

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ *      https://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 fails to start.
+ *
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+public class ConsumerFailedToStartEvent extends KafkaEvent {
+
+	private static final long serialVersionUID = 1L;
+
+	/**
+	 * Construct an instance with the provided source and container.
+	 * @param source the container instance that generated the event.
+	 * @param container the container or the parent container if the container is a child.
+	 */
+	public ConsumerFailedToStartEvent(Object source, Object container) {
+		super(source, container);
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumerFailedToStartEvent [source=" + getSource() + "]";
+	}
+
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ *      https://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 has started.
+ *
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+public class ConsumerStartedEvent extends KafkaEvent {
+
+	private static final long serialVersionUID = 1L;
+
+	/**
+	 * Construct an instance with the provided source and container.
+	 * @param source the container instance that generated the event.
+	 * @param container the container or the parent container if the container is a child.
+	 */
+	public ConsumerStartedEvent(Object source, Object container) {
+		super(source, container);
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumerStartedEvent [source=" + getSource() + "]";
+	}
+
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ *      https://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 initializing.
+ *
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+public class ConsumerStartingEvent extends KafkaEvent {
+
+	private static final long serialVersionUID = 1L;
+
+	/**
+	 * Construct an instance with the provided source and container.
+	 * @param source the container instance that generated the event.
+	 * @param container the container or the parent container if the container is a child.
+	 */
+	public ConsumerStartingEvent(Object source, Object container) {
+		super(source, container);
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumerStartingEvent [source=" + getSource() + "]";
+	}
+
+}

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

@@ -107,6 +107,8 @@ public class ContainerProperties extends ConsumerProperties {
 	 */
 	public static final float DEFAULT_NO_POLL_THRESHOLD = 3f;
 
+	private static final Duration DEFAULT_CONSUMER_START_TIMEOUT = Duration.ofSeconds(30);
+
 	private final Map<String, String> micrometerTags = new HashMap<>();
 
 	/**
@@ -176,6 +178,8 @@ public class ContainerProperties extends ConsumerProperties {
 
 	private boolean micrometerEnabled = true;
 
+	private Duration consumerStartTimout = DEFAULT_CONSUMER_START_TIMEOUT;
+
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -570,6 +574,20 @@ public class ContainerProperties extends ConsumerProperties {
 		return Collections.unmodifiableMap(this.micrometerTags);
 	}
 
+	public Duration getConsumerStartTimout() {
+		return this.consumerStartTimout;
+	}
+
+	/**
+	 * Set the timeout to wait for a consumer thread to start before logging
+	 * an error. Default 30 seconds.
+	 * @param consumerStartTimout the consumer start timeout.
+	 */
+	public void setConsumerStartTimout(Duration consumerStartTimout) {
+		Assert.notNull(consumerStartTimout, "'consumerStartTimout' cannot be null");
+		this.consumerStartTimout = consumerStartTimout;
+	}
+
 	@Override
 	public String toString() {
 		return "ContainerProperties ["

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

@@ -30,6 +30,7 @@ import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -61,8 +62,11 @@ import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaResourceHolder;
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.event.ConsumerFailedToStartEvent;
 import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.event.ConsumerResumedEvent;
+import org.springframework.kafka.event.ConsumerStartedEvent;
+import org.springframework.kafka.event.ConsumerStartingEvent;
 import org.springframework.kafka.event.ConsumerStoppedEvent;
 import org.springframework.kafka.event.ConsumerStoppingEvent;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
@@ -132,16 +136,18 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 	private final TopicPartitionOffset[] topicPartitions;
 
-	private volatile ListenerConsumer listenerConsumer;
-
-	private volatile ListenableFuture<?> listenerConsumerFuture;
-
 	private String clientIdSuffix;
 
 	private Runnable emergencyStop = () -> stop(() -> {
 		// NOSONAR
 	});
 
+	private volatile ListenerConsumer listenerConsumer;
+
+	private volatile ListenableFuture<?> listenerConsumerFuture;
+
+	private volatile CountDownLatch startLatch = new CountDownLatch(1);
+
 	/**
 	 * Construct an instance with the supplied configuration properties.
 	 * @param consumerFactory the consumer factory.
@@ -320,9 +326,20 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		ListenerType listenerType = determineListenerType(listener);
 		this.listenerConsumer = new ListenerConsumer(listener, listenerType);
 		setRunning(true);
+		this.startLatch = new CountDownLatch(1);
 		this.listenerConsumerFuture = containerProperties
 				.getConsumerTaskExecutor()
 				.submitListenable(this.listenerConsumer);
+		try {
+			if (!this.startLatch.await(containerProperties.getConsumerStartTimout().toMillis(), TimeUnit.MILLISECONDS)) {
+				this.logger.error("Consumer thread failed to start - does the configured task executor "
+						+ "have enough threads to support all containers and concurrency?");
+				publishConsumerFailedToStart();
+			}
+		}
+		catch (@SuppressWarnings("unused") InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
 	}
 
 	private void checkAckMode(ContainerProperties containerProperties) {
@@ -406,6 +423,25 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		}
 	}
 
+	private void publishConsumerStartingEvent() {
+		this.startLatch.countDown();
+		if (getApplicationEventPublisher() != null) {
+			getApplicationEventPublisher().publishEvent(new ConsumerStartingEvent(this, this.container));
+		}
+	}
+
+	private void publishConsumerStartedEvent() {
+		if (getApplicationEventPublisher() != null) {
+			getApplicationEventPublisher().publishEvent(new ConsumerStartedEvent(this, this.container));
+		}
+	}
+
+	private void publishConsumerFailedToStart() {
+		if (getApplicationEventPublisher() != null) {
+			getApplicationEventPublisher().publishEvent(new ConsumerFailedToStartEvent(this, this.container));
+		}
+	}
+
 	@Override
 	protected AbstractMessageListenerContainer<?, ?> parentOrThis() {
 		return this.container;
@@ -833,6 +869,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 		@Override
 		public void run() {
+			publishConsumerStartingEvent();
 			this.consumerThread = Thread.currentThread();
 			if (this.consumerSeekAwareListener != null) {
 				this.consumerSeekAwareListener.registerSeekCallback(this);
@@ -841,6 +878,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			this.count = 0;
 			this.last = System.currentTimeMillis();
 			initAssignedPartitions();
+			publishConsumerStartedEvent();
 			while (isRunning()) {
 				try {
 					pollAndInvoke();

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

@@ -18,17 +18,22 @@ package org.springframework.kafka.listener;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -45,7 +50,11 @@ import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.event.ConsumerFailedToStartEvent;
+import org.springframework.kafka.event.ConsumerStartedEvent;
+import org.springframework.kafka.event.ConsumerStartingEvent;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 /**
  * @author Gary Russell
@@ -54,6 +63,53 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
  */
 public class ConcurrentMessageListenerContainerMockTests {
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void testThreadStarvation() throws InterruptedException {
+		ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
+		final Consumer consumer = mock(Consumer.class);
+		Set<String> consumerThreads = ConcurrentHashMap.newKeySet();
+		CountDownLatch latch = new CountDownLatch(2);
+		willAnswer(invocation -> {
+			consumerThreads.add(Thread.currentThread().getName());
+			latch.countDown();
+			Thread.sleep(50);
+			return new ConsumerRecords<>(Collections.emptyMap());
+		}).given(consumer).poll(any());
+		given(consumerFactory.createConsumer(anyString(), anyString(), anyString(),
+				eq(KafkaTestUtils.defaultPropertyOverrides())))
+						.willReturn(consumer);
+		ContainerProperties containerProperties = new ContainerProperties("foo");
+		containerProperties.setGroupId("grp");
+		containerProperties.setMessageListener((MessageListener) record -> { });
+		containerProperties.setMissingTopicsFatal(false);
+		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
+		exec.setCorePoolSize(1);
+		exec.afterPropertiesSet();
+		containerProperties.setConsumerTaskExecutor(exec);
+		containerProperties.setConsumerStartTimout(Duration.ofMillis(50));
+		ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(consumerFactory,
+				containerProperties);
+		container.setConcurrency(2);
+		CountDownLatch startedLatch = new CountDownLatch(2);
+		CountDownLatch failedLatch = new CountDownLatch(1);
+		container.setApplicationEventPublisher(event -> {
+			if (event instanceof ConsumerStartingEvent || event instanceof ConsumerStartedEvent) {
+				startedLatch.countDown();
+			}
+			else if (event instanceof ConsumerFailedToStartEvent) {
+				failedLatch.countDown();
+			}
+		});
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(consumerThreads).hasSize(1);
+		assertThat(startedLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(failedLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		exec.destroy();
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	void testCorrectContainerForConsumerError() throws InterruptedException {

src/reference/asciidoc/kafka.adoc

@@ -1813,6 +1813,23 @@ public SeekToCurrentErrorHandler eh() {
 
 However, see the note at the beginning of this section; you can avoid using the `RetryTemplate` altogether.
 
+[[events]]
+===== Listener Consumer Lifecycle Events
+
+The following events are published when containers are started and stopped:
+
+* `ConsumerStartingEvent` - published when a consumer thread is first started, before it starts polling.
+* `ConsumerStartedEvent` - published when a consumer is about to start polling.
+* `ConsumerFailedToStartEvent` - published if no `ConsumerStartingEvent` is published within the `consumerStartTimeout` container property.
+This event might signal that the configured task executor has insufficient threads to support the containers it is used in and their concurrency.
+An error message is also logged when this condition occurs.
+* `IdleContainerEvent` - discussed in <<idle-containers>>.
+* `NonResponsiveConsumerEvent` - discussed in <<idle-containers>>.
+* `ConsumerPausedEvent` - discussed in <<pause-resume>>.
+* `ConsumerResumedEvent` - discussed in <<pause-resume>>.
+* `ConsumerStoppingEvent` - published when a consumer begins to stop.
+* `ConsumerStartedEvent` - published when a consumer is stopped.
+
 [[idle-containers]]
 ===== Detecting Idle and Non-Responsive Consumers
 

src/reference/asciidoc/whats-new.adoc

@@ -61,6 +61,9 @@ See <<committing-offsets>> for more information.
 Listener performance can now be monitored using Micrometer `Timer` s.
 See <<micrometer>> for more information.
 
+The containers now publish additional consumer lifecyle events relating to startup.
+See <<events>> for more information.
+
 ==== ErrorHandler Changes
 
 The `SeekToCurrentErrorHandler` now treats certain exceptions as fatal and disables retry for those, invoking the recoverer on first failure.