From 39bf44810668e362cdd6ab993da82b06b252edae Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 29 Nov 2017 09:15:03 -0500 Subject: [PATCH] GH-475 Fix Listener Container SmartLifecycle phase Fixes https://github.com/spring-projects/spring-kafka/issues/475 Previously, containers were started in phase 0; thy should be started in a late phase. __cherry-pick to 2.0.x, 1.3.x__ --- .../config/KafkaListenerEndpointRegistry.java | 12 ++++--- .../AbstractMessageListenerContainer.java | 7 +++- src/reference/asciidoc/kafka.adoc | 33 +++++++++++++++++++ 3 files changed, 46 insertions(+), 6 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java index a9dda910..a11e36cb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java @@ -38,6 +38,7 @@ import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.SmartLifecycle; import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -72,7 +73,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec private final Map listenerContainers = new ConcurrentHashMap(); - private int phase = Integer.MAX_VALUE; + private int phase = AbstractMessageListenerContainer.DEFAULT_PHASE; private ConfigurableApplicationContext applicationContext; @@ -191,10 +192,11 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec } int containerPhase = listenerContainer.getPhase(); - if (containerPhase < Integer.MAX_VALUE) { // a custom phase value - if (this.phase < Integer.MAX_VALUE && this.phase != containerPhase) { - throw new IllegalStateException("Encountered phase mismatch between container factory definitions: " + - this.phase + " vs " + containerPhase); + if (listenerContainer.isAutoStartup() && + containerPhase != AbstractMessageListenerContainer.DEFAULT_PHASE) { // a custom phase value + if (this.phase != AbstractMessageListenerContainer.DEFAULT_PHASE && this.phase != containerPhase) { + throw new IllegalStateException("Encountered phase mismatch between container " + + "factory definitions: " + this.phase + " vs " + containerPhase); } this.phase = listenerContainer.getPhase(); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index 66d9c217..28973b8b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -46,6 +46,11 @@ import org.springframework.util.Assert; public abstract class AbstractMessageListenerContainer implements MessageListenerContainer, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle { + /** + * The default {@link SmartLifecycle} phase for listener containers {@value #DEFAULT_PHASE}. + */ + public static final int DEFAULT_PHASE = Integer.MAX_VALUE - 100; // late phase + protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR /** @@ -109,7 +114,7 @@ public abstract class AbstractMessageListenerContainer private boolean autoStartup = true; - private int phase = 0; + private int phase = DEFAULT_PHASE; private volatile boolean running = false; diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index b2ef6060..34de965f 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -565,6 +565,14 @@ public interface Acknowledgment { This gives the listener control over when offsets are committed. +[[container-auto-startup]] +====== Listener Container Auto Startup + +The listener containers implement `SmartLifecycle` and `autoStartup` is `true` by default; the containers are started in a late phase (`Integer.MAX-VALUE - 100`). +Other components that implement `SmartLifecycle`, that handle data from listeners, should be started in an earlier phase. +The `- 100` leaves room for later phases to enable components to be auto-started after the containers. + + [[kafka-listener-annotation]] ===== @KafkaListener Annotation @@ -779,6 +787,31 @@ static class MultiListenerBean { } ---- +[[kafkalistener-lifecycle]] +===== @KafkaListener Lifecycle Management + +The listener containers created for `@KafkaListener` annotations are not beans in the application context. +Instead, they are registered with an infrastructure bean of type `KafkaListenerEndpointRegistry`. +This bean manages the containers' lifecycles; it will auto-start any containers that have `autoStartup` set to `true`. +All containers created by all container factories must be in the same `phase` - see <> for more information. +You can manage the lifecycle programmatically using the registry; starting/stopping the registry will start/stop all the registered containers. +Or, you can get a reference to an individual container using its `id` attribute; you can set `autoStartup` on the annotation, which will override the default setting configured into the container factory. + +[source, java] +---- +@Autowired +private KafkaListenerEndpointRegistry registry; + +... + +@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false") +public void listen(...) { ... } + +... + + registry.getListenerContainer("myContainer").start(); +---- + [[rebalance-listeners]] ===== Rebalance Listeners