From 68405e9f722b48cd625fabdf8b7934149310ea7d Mon Sep 17 00:00:00 2001 From: Hyun Lee Date: Thu, 24 Sep 2026 10:31:21 -0500 Subject: [PATCH] Add unregisterListenerContainer to listener endpoint registry PulsarListenerEndpointRegistry supports registering listener containers at runtime via registerListenerContainer, but offers no way to remove them. Applications that recreate listeners dynamically therefore cannot reuse endpoint ids, and the registry keeps growing with stale containers. Add GenericListenerEndpointRegistry#unregisterListenerContainer(String), mirroring KafkaListenerEndpointRegistry in Spring for Apache Kafka. The method only removes the container from the registry and returns it; it does not stop or destroy the container, which remains the caller's responsibility. Resolves #1552 Signed-off-by: Hyun Lee --- .../GenericListenerEndpointRegistry.java | 18 +++++ .../PulsarListenerEndpointRegistryTests.java | 71 +++++++++++++++++++ 2 files changed, 89 insertions(+) create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistryTests.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..5dd79cd90 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> implements PulsarListenerContainerRegistry, DisposableBean, SmartLifecycle, ApplicationContextAware, @@ -141,6 +142,23 @@ public void registerListenerContainer(E endpoint, ListenerContainerFactory + * IMPORTANT: this method only removes the container from the registry. It does not + * stop or destroy the container; call {@link MessageListenerContainer#stop()} or + * {@link MessageListenerContainer#destroy()} before or after calling this method to + * shut it down. + * @param id the id of the container + * @return the removed container or {@code null} if no container was registered with + * that id + * @since 2.0.8 + */ + @Nullable public C unregisterListenerContainer(String id) { + Assert.hasText(id, "Container identifier must not be empty"); + return this.listenerContainers.remove(id); + } + protected C createListenerContainer(E endpoint, ListenerContainerFactory factory) { C listenerContainer = factory.createRegisteredContainer(endpoint); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistryTests.java new file mode 100644 index 000000000..b46f1c298 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/config/PulsarListenerEndpointRegistryTests.java @@ -0,0 +1,71 @@ +/* + * 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.assertj.core.api.Assertions.assertThat; +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.pulsar.listener.ConcurrentPulsarMessageListenerContainer; + +/** + * Unit tests for {@link PulsarListenerEndpointRegistry}. + * + * @author Hyun Lee + */ +class PulsarListenerEndpointRegistryTests { + + private final PulsarListenerEndpointRegistry registry = new PulsarListenerEndpointRegistry(); + + @Test + void unregisterListenerContainerRemovesContainerWithoutStoppingIt() { + var container = registerContainer("foo"); + assertThat(this.registry.unregisterListenerContainer("foo")).isSameAs(container); + assertThat(this.registry.getListenerContainer("foo")).isNull(); + assertThat(this.registry.getListenerContainerIds()).isEmpty(); + verify(container, never()).stop(); + verify(container, never()).destroy(); + } + + @Test + void unregisterListenerContainerReturnsNullForUnknownId() { + assertThat(this.registry.unregisterListenerContainer("foo")).isNull(); + } + + @Test + void endpointIdCanBeReusedAfterUnregister() { + registerContainer("foo"); + this.registry.unregisterListenerContainer("foo"); + var newContainer = registerContainer("foo"); + assertThat(this.registry.getListenerContainer("foo")).isSameAs(newContainer); + } + + private ConcurrentPulsarMessageListenerContainer registerContainer(String id) { + var endpoint = mock(PulsarListenerEndpoint.class); + when(endpoint.getId()).thenReturn(id); + ConcurrentPulsarMessageListenerContainer container = mock(); + var factory = mock(PulsarListenerContainerFactory.class); + when(factory.createRegisteredContainer(endpoint)).thenReturn(container); + this.registry.registerListenerContainer(endpoint, factory); + return container; + } + +}