Add common container factory interface (#829)

This introduces a common interface for all message listener containers.
Prior to this commit, the reader and listener containers had no common
abstraction. This is needed to introduce a generic container factory
customizer in Spring Boot.

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -24,6 +24,17 @@ When using Spring Boot the `PulsarTopicBuilder` is now a registered bean that is
 Therefore, if you are using Spring Boot, you can simply inject the builder where needed.
 Otherwise, use one of the `PulsarTopicBuilder` constructors directly.
 
+==== Listener/ReaderContainerFactory
+The `PulsarContainerFactory` common interface was introduced to bridge the gap between listener and reader container factories.
+As part of this, the following APIs were deprecated, copied, and renamed:
+
+- `ListenerContainerFactory#createListenerContainer` replaced with `ListenerContainerFactory#createRegisteredContainer`
+
+- `ReaderContainerFactory#createReaderContainer(E endpoint)` replaced with `ReaderContainerFactory#createRegisteredContainer`
+
+- `ReaderContainerFactory#createReaderContainer(String... topics)` replaced with `ReaderContainerFactory#createContainer`
+
+
 === Breaking Changes
 
 ==== PulsarTopic#<init>

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -139,25 +139,21 @@ public class DefaultReactivePulsarListenerContainerFactory<T> implements Reactiv
 		return new DefaultReactivePulsarMessageListenerContainer<>(this.getConsumerFactory(), containerProps);
 	}
 
+	@SuppressWarnings("rawtypes")
 	@Override
-	public DefaultReactivePulsarMessageListenerContainer<T> createListenerContainer(
+	public DefaultReactivePulsarMessageListenerContainer<T> createRegisteredContainer(
 			ReactivePulsarListenerEndpoint<T> endpoint) {
-		DefaultReactivePulsarMessageListenerContainer<T> instance = createContainerInstance(endpoint);
-		if (endpoint instanceof AbstractReactivePulsarListenerEndpoint) {
-			configureEndpoint((AbstractReactivePulsarListenerEndpoint<T>) endpoint);
+		var instance = createContainerInstance(endpoint);
+		if (endpoint instanceof AbstractReactivePulsarListenerEndpoint abstractReactiveEndpoint) {
+			if (abstractReactiveEndpoint.getFluxListener() == null) {
+				JavaUtils.INSTANCE.acceptIfNotNull(this.fluxListener, abstractReactiveEndpoint::setFluxListener);
+			}
 		}
-
 		endpoint.setupListenerContainer(instance, this.messageConverter);
 		initializeContainer(instance, endpoint);
 		return instance;
 	}
 
-	private void configureEndpoint(AbstractReactivePulsarListenerEndpoint<T> aplEndpoint) {
-		if (aplEndpoint.getFluxListener() == null) {
-			JavaUtils.INSTANCE.acceptIfNotNull(this.fluxListener, aplEndpoint::setFluxListener);
-		}
-	}
-
 	@Override
 	public DefaultReactivePulsarMessageListenerContainer<T> createContainer(String... topics) {
 		ReactivePulsarListenerEndpoint<T> endpoint = new ReactivePulsarListenerEndpoint<>() {
@@ -168,7 +164,7 @@ public class DefaultReactivePulsarListenerContainerFactory<T> implements Reactiv
 			}
 
 		};
-		DefaultReactivePulsarMessageListenerContainer<T> container = createContainerInstance(endpoint);
+		var container = createContainerInstance(endpoint);
 		initializeContainer(container, endpoint);
 		return container;
 	}

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactoryTests.java

@@ -18,6 +18,8 @@ package org.springframework.pulsar.reactive.config;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -32,6 +34,18 @@ import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerPrope
  */
 class DefaultReactivePulsarListenerContainerFactoryTests {
 
+	@SuppressWarnings({ "removal", "unchecked" })
+	@Test
+	void deprecatedCreateListenerContainerCallsReplacementApi() {
+		var containerFactory = spy(new DefaultReactivePulsarListenerContainerFactory<String>(
+				mock(ReactivePulsarConsumerFactory.class), new ReactivePulsarContainerProperties<>()));
+		var endpoint = mock(ReactivePulsarListenerEndpoint.class);
+		when(endpoint.getConcurrency()).thenReturn(1);
+		var createdContainer = containerFactory.createListenerContainer(endpoint);
+		assertThat(createdContainer).isNotNull();
+		verify(containerFactory).createRegisteredContainer(endpoint);
+	}
+
 	@SuppressWarnings("unchecked")
 	@Nested
 	class SubscriptionTypeFrom {
@@ -44,7 +58,7 @@ class DefaultReactivePulsarListenerContainerFactoryTests {
 					mock(ReactivePulsarConsumerFactory.class), factoryProps);
 			var endpoint = mock(ReactivePulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
 				.isEqualTo(SubscriptionType.Shared);
 		}
@@ -58,7 +72,7 @@ class DefaultReactivePulsarListenerContainerFactoryTests {
 			var endpoint = mock(ReactivePulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
 			when(endpoint.getSubscriptionType()).thenReturn(SubscriptionType.Failover);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
 				.isEqualTo(SubscriptionType.Failover);
 		}
@@ -70,7 +84,7 @@ class DefaultReactivePulsarListenerContainerFactoryTests {
 					mock(ReactivePulsarConsumerFactory.class), factoryProps);
 			var endpoint = mock(ReactivePulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
 				.isEqualTo(SubscriptionType.Exclusive);
 
@@ -90,7 +104,7 @@ class DefaultReactivePulsarListenerContainerFactoryTests {
 					mock(ReactivePulsarConsumerFactory.class), factoryProps);
 			var endpoint = mock(ReactivePulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionName())
 				.isEqualTo("my-factory-subscription");
 		}
@@ -104,7 +118,7 @@ class DefaultReactivePulsarListenerContainerFactoryTests {
 			var endpoint = mock(ReactivePulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
 			when(endpoint.getSubscriptionName()).thenReturn("my-endpoint-subscription");
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionName())
 				.isEqualTo("my-endpoint-subscription");
 		}
@@ -117,10 +131,10 @@ class DefaultReactivePulsarListenerContainerFactoryTests {
 			var endpoint = mock(ReactivePulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
 
-			var container1 = containerFactory.createListenerContainer(endpoint);
+			var container1 = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(container1.getContainerProperties().getSubscriptionName())
 				.startsWith("org.springframework.Pulsar.ReactivePulsarListenerEndpointContainer#");
-			var container2 = containerFactory.createListenerContainer(endpoint);
+			var container2 = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(container2.getContainerProperties().getSubscriptionName())
 				.startsWith("org.springframework.Pulsar.ReactivePulsarListenerEndpointContainer#");
 			assertThat(container1.getContainerProperties().getSubscriptionName())

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerSpelTests.java

@@ -171,9 +171,9 @@ class ReactivePulsarListenerSpelTests extends ReactivePulsarListenerTestsBase {
 		@Test
 		void containerFactoryDerivedFromAttribute(
 				@Autowired ReactivePulsarListenerContainerFactory<String> containerFactory) {
-			verify(containerFactory).createListenerContainer(argThat(endpoint -> endpoint.getId().equals("foo")));
-			verify(containerFactory).createListenerContainer(argThat(endpoint -> endpoint.getId().equals("bar")));
-			verify(containerFactory).createListenerContainer(argThat(endpoint -> endpoint.getId().equals("zaa")));
+			verify(containerFactory).createRegisteredContainer(argThat(endpoint -> endpoint.getId().equals("foo")));
+			verify(containerFactory).createRegisteredContainer(argThat(endpoint -> endpoint.getId().equals("bar")));
+			verify(containerFactory).createRegisteredContainer(argThat(endpoint -> endpoint.getId().equals("zaa")));
 		}
 
 		@EnablePulsar

spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -106,7 +106,7 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
 
 	@SuppressWarnings("unchecked")
 	@Override
-	public C createListenerContainer(PulsarListenerEndpoint endpoint) {
+	public C createRegisteredContainer(PulsarListenerEndpoint endpoint) {
 		C instance = createContainerInstance(endpoint);
 		JavaUtils.INSTANCE.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
 		if (endpoint instanceof AbstractPulsarListenerEndpoint) {

spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarReaderContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2024 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.
@@ -99,7 +99,7 @@ public abstract class AbstractPulsarReaderContainerFactory<C extends AbstractPul
 
 	@SuppressWarnings("unchecked")
 	@Override
-	public C createReaderContainer(PulsarReaderEndpoint<PulsarMessageReaderContainer> endpoint) {
+	public C createRegisteredContainer(PulsarReaderEndpoint<PulsarMessageReaderContainer> endpoint) {
 		C instance = createContainerInstance(endpoint);
 		JavaUtils.INSTANCE.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
 		if (endpoint instanceof AbstractPulsarReaderEndpoint) {

spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarReaderContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2024 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.
@@ -56,7 +56,7 @@ public class DefaultPulsarReaderContainerFactory<T>
 	}
 
 	@Override
-	public DefaultPulsarMessageReaderContainer<T> createReaderContainer(String... topics) {
+	public DefaultPulsarMessageReaderContainer<T> createContainer(String... topics) {
 		// TODO
 		return null;
 	}

spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericListenerEndpointRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -138,7 +138,7 @@ public class GenericListenerEndpointRegistry<C extends MessageListenerContainer,
 
 	protected C createListenerContainer(E endpoint, ListenerContainerFactory<? extends C, E> factory) {
 
-		C listenerContainer = factory.createListenerContainer(endpoint);
+		C listenerContainer = factory.createRegisteredContainer(endpoint);
 
 		if (listenerContainer instanceof InitializingBean) {
 			try {

spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericReaderEndpointRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2024 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.
@@ -136,7 +136,7 @@ public class GenericReaderEndpointRegistry<C extends PulsarMessageReaderContaine
 	}
 
 	protected C createReaderContainer(E endpoint, ReaderContainerFactory<? extends C, E> factory) {
-		C readerContainer = factory.createReaderContainer(endpoint);
+		C readerContainer = factory.createRegisteredContainer(endpoint);
 		if (readerContainer instanceof InitializingBean) {
 			try {
 				((InitializingBean) readerContainer).afterPropertiesSet();

spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -16,7 +16,6 @@
 
 package org.springframework.pulsar.config;
 
-import org.springframework.pulsar.annotation.PulsarListener;
 import org.springframework.pulsar.listener.MessageListenerContainer;
 
 /**
@@ -26,24 +25,22 @@ import org.springframework.pulsar.listener.MessageListenerContainer;
  * @param <E> listener endpoint type.
  * @author Soby Chacko
  * @author Christophe Bornet
+ * @author Chris Bono
  */
-public interface ListenerContainerFactory<C extends MessageListenerContainer, E extends ListenerEndpoint<C>> {
+public interface ListenerContainerFactory<C extends MessageListenerContainer, E extends ListenerEndpoint<C>>
+		extends PulsarContainerFactory<C, E> {
 
 	/**
 	 * Create a {@link MessageListenerContainer} for the given {@link ListenerEndpoint}.
 	 * Containers created using this method are added to the listener endpoint registry.
 	 * @param endpoint the endpoint to configure
 	 * @return the created container
+	 * @deprecated since 1.2.0 for removal in 1.4.0 in favor of
+	 * {@link PulsarContainerFactory#createRegisteredContainer}
 	 */
-	C createListenerContainer(E endpoint);
-
-	/**
-	 * Create and configure a container without a listener; used to create containers that
-	 * are not used for {@link PulsarListener} annotations. Containers created using this
-	 * method are not added to the listener endpoint registry.
-	 * @param topics the topics.
-	 * @return the container.
-	 */
-	C createContainer(String... topics);
+	@Deprecated(since = "1.2.0", forRemoval = true)
+	default C createListenerContainer(E endpoint) {
+		return createRegisteredContainer(endpoint);
+	}
 
 }

spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -41,7 +41,7 @@ public interface ListenerEndpoint<C extends MessageListenerContainer> {
 	 * Return the id of this endpoint.
 	 * @return the id of this endpoint. The id can be further qualified when the endpoint
 	 * is resolved against its actual listener container.
-	 * @see ListenerContainerFactory#createListenerContainer
+	 * @see ListenerContainerFactory#createRegisteredContainer
 	 */
 	@Nullable
 	default String getId() {

spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarContainerFactory.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2022-2024 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;
+
+/**
+ * Factory for Pulsar message listener containers.
+ *
+ * @param <C> message container
+ * @param <E> message listener endpoint
+ * @author Chris Bono
+ * @since 1.2.0
+ */
+public interface PulsarContainerFactory<C, E> {
+
+	/**
+	 * Create a message listener container for the given endpoint. Containers created
+	 * using this method are added to the listener endpoint registry.
+	 * @param endpoint the endpoint to configure
+	 * @return the created container
+	 */
+	C createRegisteredContainer(E endpoint);
+
+	/**
+	 * Create and configure a container without a listener. Containers created using this
+	 * method are not added to the listener endpoint registry.
+	 * @param topics the topics.
+	 * @return the container.
+	 */
+	C createContainer(String... topics);
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarReaderEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2024 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.
@@ -39,7 +39,7 @@ public interface PulsarReaderEndpoint<C extends PulsarMessageReaderContainer> {
 	 * Return the id of this endpoint.
 	 * @return the id of this endpoint. The id can be further qualified when the endpoint
 	 * is resolved against its actual listener container.
-	 * @see ListenerContainerFactory#createListenerContainer
+	 * @see ListenerContainerFactory#createRegisteredContainer
 	 */
 	@Nullable
 	String getId();

spring-pulsar/src/main/java/org/springframework/pulsar/config/ReaderContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2023-2024 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.
@@ -24,11 +24,34 @@ import org.springframework.pulsar.reader.PulsarMessageReaderContainer;
  * @param <C> Container type
  * @param <E> Endpoint type
  * @author Soby Chacko
+ * @author Chris Bono
  */
-public interface ReaderContainerFactory<C extends PulsarMessageReaderContainer, E extends PulsarReaderEndpoint<C>> {
+public interface ReaderContainerFactory<C extends PulsarMessageReaderContainer, E extends PulsarReaderEndpoint<C>>
+		extends PulsarContainerFactory<C, E> {
 
-	C createReaderContainer(E endpoint);
+	/**
+	 * Create a message reader container for the given endpoint and register the container
+	 * with the listener endpoint registry.
+	 * @param endpoint reader endpoint
+	 * @return the created container
+	 * @deprecated since 1.2.0 for removal in 1.4.0 in favor of
+	 * {@link PulsarContainerFactory#createRegisteredContainer}
+	 */
+	@Deprecated(since = "1.2.0", forRemoval = true)
+	default C createReaderContainer(E endpoint) {
+		return createRegisteredContainer(endpoint);
+	}
 
-	C createReaderContainer(String... topics);
+	/**
+	 * Create a message reader container for the given endpoint.
+	 * @param topics the topics to read from
+	 * @return the created container
+	 * @deprecated since 1.2.0 for removal in 1.4.0 in favor of
+	 * {@link PulsarContainerFactory#createContainer}
+	 */
+	@Deprecated(since = "1.2.0", forRemoval = true)
+	default C createReaderContainer(String... topics) {
+		return createContainer(topics);
+	}
 
 }

spring-pulsar/src/test/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryTests.java

@@ -18,6 +18,8 @@ package org.springframework.pulsar.config;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -32,6 +34,18 @@ import org.springframework.pulsar.listener.PulsarContainerProperties;
  */
 class ConcurrentPulsarListenerContainerFactoryTests {
 
+	@SuppressWarnings({ "removal", "unchecked" })
+	@Test
+	void deprecatedCreateListenerContainerCallsReplacementApi() {
+		var containerFactory = spy(new ConcurrentPulsarListenerContainerFactory<String>(
+				mock(PulsarConsumerFactory.class), new PulsarContainerProperties()));
+		var endpoint = mock(PulsarListenerEndpoint.class);
+		when(endpoint.getConcurrency()).thenReturn(1);
+		var createdContainer = containerFactory.createListenerContainer(endpoint);
+		assertThat(createdContainer).isNotNull();
+		verify(containerFactory).createRegisteredContainer(endpoint);
+	}
+
 	@SuppressWarnings("unchecked")
 	@Nested
 	class SubscriptionTypeFrom {
@@ -44,7 +58,7 @@ class ConcurrentPulsarListenerContainerFactoryTests {
 					mock(PulsarConsumerFactory.class), factoryProps);
 			var endpoint = mock(PulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
 				.isEqualTo(SubscriptionType.Shared);
 		}
@@ -58,7 +72,7 @@ class ConcurrentPulsarListenerContainerFactoryTests {
 			var endpoint = mock(PulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
 			when(endpoint.getSubscriptionType()).thenReturn(SubscriptionType.Failover);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
 				.isEqualTo(SubscriptionType.Failover);
 		}
@@ -70,7 +84,7 @@ class ConcurrentPulsarListenerContainerFactoryTests {
 					mock(PulsarConsumerFactory.class), factoryProps);
 			var endpoint = mock(PulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
 				.isEqualTo(SubscriptionType.Exclusive);
 		}
@@ -89,7 +103,7 @@ class ConcurrentPulsarListenerContainerFactoryTests {
 					mock(PulsarConsumerFactory.class), factoryProps);
 			var endpoint = mock(PulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionName())
 				.isEqualTo("my-factory-subscription");
 		}
@@ -103,7 +117,7 @@ class ConcurrentPulsarListenerContainerFactoryTests {
 			var endpoint = mock(PulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
 			when(endpoint.getSubscriptionName()).thenReturn("my-endpoint-subscription");
-			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			var createdContainer = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(createdContainer.getContainerProperties().getSubscriptionName())
 				.isEqualTo("my-endpoint-subscription");
 		}
@@ -116,10 +130,10 @@ class ConcurrentPulsarListenerContainerFactoryTests {
 			var endpoint = mock(PulsarListenerEndpoint.class);
 			when(endpoint.getConcurrency()).thenReturn(1);
 
-			var container1 = containerFactory.createListenerContainer(endpoint);
+			var container1 = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(container1.getContainerProperties().getSubscriptionName())
 				.startsWith("org.springframework.Pulsar.PulsarListenerEndpointContainer#");
-			var container2 = containerFactory.createListenerContainer(endpoint);
+			var container2 = containerFactory.createRegisteredContainer(endpoint);
 			assertThat(container2.getContainerProperties().getSubscriptionName())
 				.startsWith("org.springframework.Pulsar.PulsarListenerEndpointContainer#");
 			assertThat(container1.getContainerProperties().getSubscriptionName())

spring-pulsar/src/test/java/org/springframework/pulsar/config/DefaultPulsarReaderContainerFactoryTests.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2023-2024 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.spy;
+import static org.mockito.Mockito.verify;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.pulsar.core.PulsarReaderFactory;
+import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
+
+/**
+ * Unit tests for {@link DefaultPulsarReaderContainerFactory}.
+ */
+class DefaultPulsarReaderContainerFactoryTests {
+
+	@SuppressWarnings({ "removal", "unchecked" })
+	@Test
+	void deprecatedCreateReaderContainerWithEndpointCallsReplacementApi() {
+		var containerFactory = spy(new DefaultPulsarReaderContainerFactory<>(mock(PulsarReaderFactory.class),
+				new PulsarReaderContainerProperties()));
+		var endpoint = mock(PulsarReaderEndpoint.class);
+		var createdContainer = containerFactory.createReaderContainer(endpoint);
+		assertThat(createdContainer).isNotNull();
+		verify(containerFactory).createRegisteredContainer(endpoint);
+	}
+
+	@SuppressWarnings({ "removal", "unchecked" })
+	@Test
+	void deprecatedCreateReaderContainerWithTopicsCallsReplacementApi() {
+		var containerFactory = spy(new DefaultPulsarReaderContainerFactory<>(mock(PulsarReaderFactory.class),
+				new PulsarReaderContainerProperties()));
+		var createdContainer = containerFactory.createReaderContainer("my-topic");
+		// reader does not implement this API - still ensure the replacement API is called
+		assertThat(createdContainer).isNull();
+		verify(containerFactory).createContainer("my-topic");
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java

@@ -70,7 +70,7 @@ public class ConcurrentPulsarMessageListenerContainerTests {
 		when(pulsarListenerEndpoint.getConcurrency()).thenReturn(1);
 
 		AbstractPulsarMessageListenerContainer<String> concurrentContainer = containerFactory
-			.createListenerContainer(pulsarListenerEndpoint);
+			.createRegisteredContainer(pulsarListenerEndpoint);
 
 		PulsarContainerProperties pulsarContainerProperties = concurrentContainer.getContainerProperties();
 		assertThat(pulsarContainerProperties.getBatchTimeoutMillis()).isEqualTo(60_000);

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerSpelTests.java

@@ -172,9 +172,9 @@ class PulsarListenerSpelTests extends PulsarListenerTestsBase {
 
 		@Test
 		void containerFactoryDerivedFromAttribute(@Autowired PulsarListenerContainerFactory containerFactory) {
-			verify(containerFactory).createListenerContainer(argThat(endpoint -> endpoint.getId().equals("foo")));
-			verify(containerFactory).createListenerContainer(argThat(endpoint -> endpoint.getId().equals("bar")));
-			verify(containerFactory).createListenerContainer(argThat(endpoint -> endpoint.getId().equals("zaa")));
+			verify(containerFactory).createRegisteredContainer(argThat(endpoint -> endpoint.getId().equals("foo")));
+			verify(containerFactory).createRegisteredContainer(argThat(endpoint -> endpoint.getId().equals("bar")));
+			verify(containerFactory).createRegisteredContainer(argThat(endpoint -> endpoint.getId().equals("zaa")));
 		}
 
 		@EnablePulsar
@@ -187,7 +187,7 @@ class PulsarListenerSpelTests extends PulsarListenerTestsBase {
 				var mockContainerFactory = mock(PulsarListenerContainerFactory.class);
 				AbstractPulsarMessageListenerContainer<?> mockContainer = mock(
 						AbstractPulsarMessageListenerContainer.class);
-				when(mockContainerFactory.createListenerContainer(any(PulsarListenerEndpoint.class)))
+				when(mockContainerFactory.createRegisteredContainer(any(PulsarListenerEndpoint.class)))
 					.thenReturn(mockContainer);
 				return mockContainerFactory;
 			}