diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerFailedToStartEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerFailedToStartEvent.java new file mode 100644 index 00000000..43c36ba2 --- /dev/null +++ b/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() + "]"; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStartedEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStartedEvent.java new file mode 100644 index 00000000..a3a0b0a8 --- /dev/null +++ b/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() + "]"; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStartingEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStartingEvent.java new file mode 100644 index 00000000..c38acd6b --- /dev/null +++ b/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() + "]"; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java index 3e3ad467..353b8fa8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java +++ b/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 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 [" diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index f3acd83c..7210a37e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/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 // 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 // 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 // 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 // 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 // NOSONAR line count this.count = 0; this.last = System.currentTimeMillis(); initAssignedPartitions(); + publishConsumerStartedEvent(); while (isRunning()) { try { pollAndInvoke(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java index 955a4ecf..641d2dc4 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java +++ b/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 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 { diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 914be7a3..16071364 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/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 <>. +* `NonResponsiveConsumerEvent` - discussed in <>. +* `ConsumerPausedEvent` - discussed in <>. +* `ConsumerResumedEvent` - discussed in <>. +* `ConsumerStoppingEvent` - published when a consumer begins to stop. +* `ConsumerStartedEvent` - published when a consumer is stopped. + [[idle-containers]] ===== Detecting Idle and Non-Responsive Consumers diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 6ace8c7a..3e1128e7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -61,6 +61,9 @@ See <> for more information. Listener performance can now be monitored using Micrometer `Timer` s. See <> for more information. +The containers now publish additional consumer lifecyle events relating to startup. +See <> for more information. + ==== ErrorHandler Changes The `SeekToCurrentErrorHandler` now treats certain exceptions as fatal and disables retry for those, invoking the recoverer on first failure.