GH-2297: Add KLERegistry.alwaysStartAfterRefresh

Resolves #2297

Add an option to apply `autoStartup` semantics to late registrations.

**cherry-pick to 2.9.x, 2.8.x**

Author: garyrussell

Diff for: spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -1953,6 +1953,10 @@ A collection of managed containers can be obtained by calling the registry's `ge
 Version 2.2.5 added a convenience method `getAllListenerContainers()`, which returns a collection of all containers, including those managed by the registry and those declared as beans.
 The collection returned will include any prototype beans that have been initialized, but it will not initialize any lazy bean declarations.
 
+IMPORTANT: Endpoints registered after the application context has been refreshed will start immediately, regardless of their `autoStartup` property, to comply with the `SmartLifecycle` contract, where `autoStartup` is only considered during application context initialization.
+An example of late registration is a bean with a `@KafkaListener` in prototype scope where an instance is created after the context is initialized.
+Starting with version 2.8.7, you can set the registry's `alwaysStartAfterRefresh` property to `false` and then the container's `autoStartup` property will define whether or not the container is started.
+
 [[kafka-validation]]
 ===== `@KafkaListener` `@Payload` Validation
 

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

@@ -83,6 +83,8 @@ public class KafkaListenerEndpointRegistry implements ListenerContainerRegistry,
 
 	private boolean contextRefreshed;
 
+	private boolean alwaysStartAfterRefresh = true;
+
 	private volatile boolean running;
 
 	@Override
@@ -107,6 +109,20 @@ public MessageListenerContainer getListenerContainer(String id) {
 		return this.listenerContainers.get(id);
 	}
 
+	/**
+	 * By default, containers registered for endpoints after the context is refreshed
+	 * are immediately started, regardless of their autoStartup property, to comply with
+	 * the {@link SmartLifecycle} contract, where autoStartup is only considered during
+	 * context initialization. Set to false to apply the autoStartup property, even for
+	 * late endpoint binding. If this is called after the context is refreshed, it will
+	 * apply to any endpoints registered after that call.
+	 * @param alwaysStartAfterRefresh false to apply the property.
+	 * @since 2.8.7
+	 */
+	public void setAlwaysStartAfterRefresh(boolean alwaysStartAfterRefresh) {
+		this.alwaysStartAfterRefresh = alwaysStartAfterRefresh;
+	}
+
 	/**
 	 * Return the ids of the managed {@link MessageListenerContainer} instance(s).
 	 * @return the ids.
@@ -327,7 +343,7 @@ public void onApplicationEvent(ContextRefreshedEvent event) {
 	 * @see MessageListenerContainer#isAutoStartup()
 	 */
 	private void startIfNecessary(MessageListenerContainer listenerContainer) {
-		if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
+		if ((this.contextRefreshed && this.alwaysStartAfterRefresh) || listenerContainer.isAutoStartup()) {
 			listenerContainer.start();
 		}
 	}

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

@@ -71,9 +71,12 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Role;
+import org.springframework.context.annotation.Scope;
 import org.springframework.context.event.EventListener;
 import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
 import org.springframework.core.MethodParameter;
@@ -179,7 +182,7 @@
 		"annotated25", "annotated25reply1", "annotated25reply2", "annotated26", "annotated27", "annotated28",
 		"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32", "annotated33",
 		"annotated34", "annotated35", "annotated36", "annotated37", "foo", "manualStart", "seekOnIdle",
-		"annotated38", "annotated38reply", "annotated39", "annotated40", "annotated41" })
+		"annotated38", "annotated38reply", "annotated39", "annotated40", "annotated41", "annotated42" })
 @TestPropertySource(properties = "spel.props=fetch.min.bytes=420000,max.poll.records=10")
 public class EnableKafkaIntegrationTests {
 
@@ -981,6 +984,14 @@ public void testContentConversion() throws InterruptedException {
 		assertThat(this.listener.contentFoo).isEqualTo(new Foo("bar"));
 	}
 
+	@Test
+	void proto(@Autowired ApplicationContext context) {
+		this.registry.setAlwaysStartAfterRefresh(false);
+		context.getBean(ProtoListener.class);
+		assertThat(this.registry.getListenerContainer("proto").isRunning()).isFalse();
+		this.registry.setAlwaysStartAfterRefresh(true);
+	}
+
 	@Configuration
 	@EnableKafka
 	@EnableTransactionManagement(proxyTargetClass = true)
@@ -1710,6 +1721,20 @@ String barInfo() {
 			return "info for the bar listener";
 		}
 
+		@Bean
+		@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+		ProtoListener proto() {
+			return new ProtoListener();
+		}
+
+	}
+
+	static class ProtoListener {
+
+		@KafkaListener(id = "proto", topics = "annotated-42", autoStartup = "false")
+		public void listen(String in) {
+		}
+
 	}
 
 	@Component