Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Batch receive properties not copied properly

Browse Source 
* When Boot provides properties for batch receive, they are not copied
  downstream through the ConcurrentPulsarListenerContainerFactory.
  Fixing this issue.
This commit is contained in:
Soby Chacko
2022-09-16 18:22:52 -04:00
parent 5afb4d719c
commit 9d48b8df1b
2 changed files with 47 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,10 +16,14 @@
package org.springframework.pulsar.config;
import java.util.Arrays;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.Schema;
import org.springframework.beans.BeanWrapper;
import org.springframework.beans.BeansException;
import org.springframework.beans.PropertyAccessorFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
@@ -171,6 +175,9 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
instance.setAutoStartup(this.autoStartup);
}
copyProperties(this.containerProperties, instance.getContainerProperties(),
Arrays.asList("maxNumMessages", "maxNumBytes", "batchTimeout"));
JavaUtils.INSTANCE.acceptIfNotNull(this.phase, instance::setPhase)
.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
@@ -178,4 +185,17 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
instance.getContainerProperties()::setPulsarConsumerProperties);
}
/**
* Copy a list of properties from the source object to the target.
* @param source Object source to copy from
* @param target Object target to copy to
* @param requestProperties list of properties to copy from source to target
*/
public static void copyProperties(Object source, Object target, Iterable<String> requestProperties) {
BeanWrapper wrappedSource = PropertyAccessorFactory.forBeanPropertyAccess(source);
BeanWrapper wrappedTarget = PropertyAccessorFactory.forBeanPropertyAccess(target);
requestProperties.forEach(p -> wrappedTarget.setPropertyValue(p, wrappedSource.getPropertyValue(p)));
}
}

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

@@ -38,6 +38,8 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
import org.junit.jupiter.api.Test;
import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpoint;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.util.backoff.BackOff;
@@ -47,6 +49,31 @@ import org.springframework.util.backoff.BackOff;
*/
public class ConcurrentPulsarMessageListenerContainerTests {
@Test
@SuppressWarnings("unchecked")
void createConcurrentContainerFromFactoryAndVerifyBatchReceivePolicy() {
ConcurrentPulsarListenerContainerFactory<String> factory = new ConcurrentPulsarListenerContainerFactory<>();
final PulsarConsumerFactory<Object> pulsarConsumerFactory = mock(PulsarConsumerFactory.class);
factory.setPulsarConsumerFactory(pulsarConsumerFactory);
PulsarContainerProperties containerProperties = factory.getContainerProperties();
containerProperties.setBatchTimeout(60_000);
containerProperties.setMaxNumMessages(120);
containerProperties.setMaxNumBytes(32000);
factory.setConcurrency(1);
PulsarListenerEndpoint pulsarListenerEndpoint = mock(PulsarListenerEndpoint.class);
when(pulsarListenerEndpoint.getConcurrency()).thenReturn(1);
final ConcurrentPulsarMessageListenerContainer<String> concurrentContainer = factory
.createListenerContainer(pulsarListenerEndpoint);
final PulsarContainerProperties pulsarContainerProperties = concurrentContainer.getContainerProperties();
assertThat(pulsarContainerProperties.getBatchTimeout()).isEqualTo(60_000);
assertThat(pulsarContainerProperties.getMaxNumMessages()).isEqualTo(120);
assertThat(pulsarContainerProperties.getMaxNumBytes()).isEqualTo(32_000);
}
@Test
void deadLetterPolicyAppliedOnChildContainer() throws Exception {
PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared);