From 5e040ac4a28ca8fcab3b4ccabe6b0b1c100155bf Mon Sep 17 00:00:00 2001
From: Hyun Lee <dlwhdugs4147@gmail.com>
Date: Sat, 26 Sep 2026 22:09:22 -0500
Subject: [PATCH] Honor startImmediately when registering a listener container

GenericListenerEndpointRegistry#registerListenerContainer(endpoint,
factory, startImmediately) ignored the startImmediately flag, so a
container registered at runtime was never started, even with
startImmediately set to true. Callers had to look the container up and
start it themselves.

Start the new container right away when startImmediately is true and
either the application context has been refreshed or the container is
auto-startup, as KafkaListenerEndpointRegistry does in Spring for Apache
Kafka.

Signed-off-by: Hyun Lee <dlwhdugs4147@gmail.com>
---
 .../GenericListenerEndpointRegistry.java      |  4 +
 .../GenericListenerEndpointRegistryTests.java | 78 +++++++++++++++++++
 2 files changed, 82 insertions(+)
 create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/config/GenericListenerEndpointRegistryTests.java

diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericListenerEndpointRegistry.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericListenerEndpointRegistry.java
index 7a027b4a1..8704dc471 100644
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericListenerEndpointRegistry.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericListenerEndpointRegistry.java
@@ -60,6 +60,7 @@
  * @author Soby Chacko
  * @author Christophe Bornet
  * @author Chris Bono
+ * @author Hyun Lee
  */
 public class GenericListenerEndpointRegistry<C extends MessageListenerContainer, E extends ListenerEndpoint<C>>
 		implements PulsarListenerContainerRegistry, DisposableBean, SmartLifecycle, ApplicationContextAware,
@@ -135,6 +136,9 @@ public void registerListenerContainer(E endpoint, ListenerContainerFactory<? ext
 					"Another endpoint is already registered with id '" + id + "'");
 			C container = createListenerContainer(endpoint, factory);
 			this.listenerContainers.put(id, container);
+			if (startImmediately) {
+				startIfNecessary(container);
+			}
 		}
 		finally {
 			this.containersLock.unlock();
diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/config/GenericListenerEndpointRegistryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/config/GenericListenerEndpointRegistryTests.java
new file mode 100644
index 000000000..54f9fa6ff
--- /dev/null
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/config/GenericListenerEndpointRegistryTests.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2026-present 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.pulsar.config;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.pulsar.listener.ConcurrentPulsarMessageListenerContainer;
+
+/**
+ * Unit tests for {@link GenericListenerEndpointRegistry}.
+ *
+ * @author Hyun Lee
+ */
+class GenericListenerEndpointRegistryTests {
+
+	private final PulsarListenerEndpointRegistry registry = new PulsarListenerEndpointRegistry();
+
+	@Test
+	void registerWithStartImmediatelyStartsAutoStartupContainer() {
+		var container = registerContainer(true, true);
+		verify(container).start();
+	}
+
+	@Test
+	void registerWithStartImmediatelyStartsContainerOnceContextIsRefreshed() {
+		var applicationContext = mock(ConfigurableApplicationContext.class);
+		this.registry.setApplicationContext(applicationContext);
+		this.registry.onApplicationEvent(new ContextRefreshedEvent(applicationContext));
+		var container = registerContainer(false, true);
+		verify(container).start();
+	}
+
+	@Test
+	void registerWithStartImmediatelyDoesNotStartContainerWithoutAutoStartupBeforeContextIsRefreshed() {
+		var container = registerContainer(false, true);
+		verify(container, never()).start();
+	}
+
+	@Test
+	void registerWithoutStartImmediatelyDoesNotStartContainer() {
+		var container = registerContainer(true, false);
+		verify(container, never()).start();
+	}
+
+	private ConcurrentPulsarMessageListenerContainer<?> registerContainer(boolean autoStartup,
+			boolean startImmediately) {
+		var endpoint = mock(PulsarListenerEndpoint.class);
+		when(endpoint.getId()).thenReturn("foo");
+		ConcurrentPulsarMessageListenerContainer<?> container = mock();
+		when(container.isAutoStartup()).thenReturn(autoStartup);
+		var factory = mock(PulsarListenerContainerFactory.class);
+		when(factory.createRegisteredContainer(endpoint)).thenReturn(container);
+		this.registry.registerListenerContainer(endpoint, factory, startImmediately);
+		return container;
+	}
+
+}