From 875ffb33826ea8a20ecd490706ee896f717b1b4d Mon Sep 17 00:00:00 2001 From: Daniel Szabo Date: Thu, 1 May 2025 14:29:59 -0500 Subject: [PATCH] Respect executor set on container props Makes sure that the ConcurrentPulsarListenerContainerFactory copies the task executor from the factory container properties to the container instance properties. Backports the fix for #1103 from https://github.com/spring-projects/spring-pulsar/commit/6dcc813576a49aec3196e4ad5f8dca23381250b4. The change was manually applied as the concurrent unit test class diverged heavily. See #1103 Signed-off-by: Daniel Szabo --- ...currentPulsarListenerContainerFactory.java | 4 +- ...ntPulsarListenerContainerFactoryTests.java | 60 +++++++++++++++++++ 2 files changed, 63 insertions(+), 1 deletion(-) create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryTests.java diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java index 5a947ea7..e149fee0 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2023 the original author or authors. + * Copyright 2022-2025 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. @@ -33,6 +33,7 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Chris Bono * @author Alexander Preuß + * @author Daniel Szabo */ public class ConcurrentPulsarListenerContainerFactory extends AbstractPulsarListenerContainerFactory, T> { @@ -74,6 +75,7 @@ public class ConcurrentPulsarListenerContainerFactory PulsarContainerProperties properties = new PulsarContainerProperties(); properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver()); properties.setTopicResolver(this.getContainerProperties().getTopicResolver()); + properties.setConsumerTaskExecutor(this.getContainerProperties().getConsumerTaskExecutor()); var parentTxnProps = this.getContainerProperties().transactions(); var childTxnProps = properties.transactions(); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryTests.java new file mode 100644 index 00000000..32384ad6 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactoryTests.java @@ -0,0 +1,60 @@ +/* + * Copyright 2025-2025 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.when; + +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.pulsar.core.PulsarConsumerFactory; +import org.springframework.pulsar.listener.PulsarContainerProperties; + +/** + * Unit tests for {@link ConcurrentPulsarListenerContainerFactory}. + * + * @author Daniel Szabo + * @author Chris Bono + */ +class ConcurrentPulsarListenerContainerFactoryTests { + + @Nested + class ConsumerTaskExecutorFrom { + + @Test + @SuppressWarnings("unchecked") + void factoryPropsUsedWhenSpecified() { + var factoryProps = new PulsarContainerProperties(); + AsyncTaskExecutor executor = mock(); + factoryProps.setConsumerTaskExecutor(executor); + var containerFactory = new ConcurrentPulsarListenerContainerFactory( + mock(PulsarConsumerFactory.class), factoryProps); + var endpoint = mock(PulsarListenerEndpoint.class); + when(endpoint.getConcurrency()).thenReturn(1); + + var container = containerFactory.createContainerInstance(endpoint); + assertThat(container.getContainerProperties()) + .extracting(PulsarContainerProperties::getConsumerTaskExecutor) + .isSameAs(executor); + } + + } + +}