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 ea1b6fb9..ce06c357 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-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. @@ -36,6 +36,7 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Chris Bono * @author Alexander Preuß + * @author Vedran Pavic */ public class ConcurrentPulsarListenerContainerFactory extends AbstractPulsarListenerContainerFactory, T> { @@ -44,8 +45,6 @@ public class ConcurrentPulsarListenerContainerFactory private static final AtomicInteger COUNTER = new AtomicInteger(); - private Integer concurrency; - public ConcurrentPulsarListenerContainerFactory(PulsarConsumerFactory consumerFactory, PulsarContainerProperties containerProperties) { super(consumerFactory, containerProperties); @@ -55,8 +54,9 @@ public class ConcurrentPulsarListenerContainerFactory * Specify the container concurrency. * @param concurrency the number of consumers to create. */ + @Deprecated(since = "1.2.0", forRemoval = true) public void setConcurrency(Integer concurrency) { - this.concurrency = concurrency; + getContainerProperties().setConcurrency(concurrency); } @Override @@ -71,7 +71,6 @@ public class ConcurrentPulsarListenerContainerFactory }; ConcurrentPulsarMessageListenerContainer container = createContainerInstance(endpoint); initializeContainer(container, endpoint); - // customizeContainer(container); return container; } @@ -130,8 +129,8 @@ public class ConcurrentPulsarListenerContainerFactory if (endpoint.getConcurrency() != null) { instance.setConcurrency(endpoint.getConcurrency()); } - else if (this.concurrency != null) { - instance.setConcurrency(this.concurrency); + else if (getContainerProperties().getConcurrency() > 0) { + instance.setConcurrency(getContainerProperties().getConcurrency()); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java index 1699744c..ccc99038 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.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. @@ -46,6 +46,7 @@ import io.micrometer.observation.ObservationRegistry; * @author Soby Chacko * @author Alexander Preuß * @author Chris Bono + * @author Vedran Pavic */ public class PulsarContainerProperties { @@ -77,6 +78,8 @@ public class PulsarContainerProperties { private AsyncTaskExecutor consumerTaskExecutor; + private int concurrency = 1; + private int maxNumMessages = -1; private int maxNumBytes = 10 * 1024 * 1024; @@ -127,6 +130,14 @@ public class PulsarContainerProperties { this.consumerTaskExecutor = consumerExecutor; } + public int getConcurrency() { + return this.concurrency; + } + + public void setConcurrency(int concurrency) { + this.concurrency = concurrency; + } + public SubscriptionType getSubscriptionType() { return this.subscriptionType; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java index e1f9d7ca..fcae1beb 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.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. @@ -63,9 +63,9 @@ public class ConcurrentPulsarMessageListenerContainerTests { containerProperties.setBatchTimeoutMillis(60_000); containerProperties.setMaxNumMessages(120); containerProperties.setMaxNumBytes(32000); + containerProperties.setConcurrency(1); ConcurrentPulsarListenerContainerFactory containerFactory = new ConcurrentPulsarListenerContainerFactory<>( consumerFactory, containerProperties); - containerFactory.setConcurrency(1); PulsarListenerEndpoint pulsarListenerEndpoint = mock(PulsarListenerEndpoint.class); when(pulsarListenerEndpoint.getConcurrency()).thenReturn(1);