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); + } + + } + +}