From db3782160e2586b3cbb3fc1e927b681bd782604e Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 16 Sep 2022 20:24:50 -0500 Subject: [PATCH] Polish "Batch receive properties not copied properly" --- .../PulsarAnnotationDrivenConfiguration.java | 2 +- .../PulsarAutoConfigurationTests.java | 2 +- ...bstractPulsarListenerContainerFactory.java | 49 ++++++------------- ...DefaultPulsarMessageListenerContainer.java | 2 +- .../listener/PulsarContainerProperties.java | 10 ++-- .../core/ConsumerAcknowledgmentTests.java | 4 +- ...ntPulsarMessageListenerContainerTests.java | 4 +- ...efaultPulsarConsumerErrorHandlerTests.java | 10 ++-- 8 files changed, 32 insertions(+), 51 deletions(-) diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java index e8a0cac5..c49f4599 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java @@ -61,7 +61,7 @@ public class PulsarAnnotationDrivenConfiguration { map.from(properties::getSchemaType).to(containerProperties::setSchemaType); map.from(properties::getAckMode).to(containerProperties::setAckMode); - map.from(properties::getBatchTimeoutMillis).to(containerProperties::setBatchTimeout); + map.from(properties::getBatchTimeoutMillis).to(containerProperties::setBatchTimeoutMillis); map.from(properties::getMaxNumBytes).to(containerProperties::setMaxNumBytes); map.from(properties::getMaxNumMessages).to(containerProperties::setMaxNumMessages); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index 8e275d98..b8410745 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -210,7 +210,7 @@ class PulsarAutoConfigurationTests { .getBean(ConcurrentPulsarListenerContainerFactory.class).extracting("containerProperties") .hasFieldOrPropertyWithValue("maxNumMessages", 10) .hasFieldOrPropertyWithValue("maxNumBytes", 101) - .hasFieldOrPropertyWithValue("batchTimeout", 50))); + .hasFieldOrPropertyWithValue("batchTimeoutMillis", 50))); } @Nested diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java index 31a4a966..cf4adf08 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java @@ -16,14 +16,10 @@ 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; @@ -136,35 +132,32 @@ public abstract class AbstractPulsarListenerContainerFactory aplEndpoint) { - if (aplEndpoint.getBatchListener() == null) { JavaUtils.INSTANCE.acceptIfNotNull(this.batchListener, aplEndpoint::setBatchListener); } } protected void initializeContainer(C instance, PulsarListenerEndpoint endpoint) { - PulsarContainerProperties properties = instance.getPulsarContainerProperties(); - // BeanUtils.copyProperties(this.containerProperties, properties, "topics", - // "messageListener", - // "batchListener", "subscriptionName", "subscriptionType", "schema"); + PulsarContainerProperties instanceProperties = instance.getPulsarContainerProperties(); - if (properties.getSchemaType() == null) { - if (this.containerProperties.getSchemaType() != null) { - properties.setSchemaType(this.containerProperties.getSchemaType()); - } + if (instanceProperties.getSchemaType() == null) { + JavaUtils.INSTANCE.acceptIfNotNull(this.containerProperties.getSchemaType(), + instanceProperties::setSchemaType); } - if (properties.getSchema() == null) { - properties.setSchema(Schema.BYTES); + + if (instanceProperties.getSchema() == null) { + instanceProperties.setSchema(Schema.BYTES); } - if (properties.getSubscriptionType() == null) { - properties.setSubscriptionType(this.containerProperties.getSubscriptionType()); + + if (instanceProperties.getSubscriptionType() == null) { + instanceProperties.setSubscriptionType(this.containerProperties.getSubscriptionType()); } if (endpoint.getAckMode() != AckMode.BATCH) { - properties.setAckMode(endpoint.getAckMode()); + instanceProperties.setAckMode(endpoint.getAckMode()); } else if (this.containerProperties.getAckMode() != AckMode.BATCH) { - properties.setAckMode(this.containerProperties.getAckMode()); + instanceProperties.setAckMode(this.containerProperties.getAckMode()); } Boolean autoStart = endpoint.getAutoStartup(); @@ -175,8 +168,9 @@ public abstract class AbstractPulsarListenerContainerFactory requestProperties) { - BeanWrapper wrappedSource = PropertyAccessorFactory.forBeanPropertyAccess(source); - BeanWrapper wrappedTarget = PropertyAccessorFactory.forBeanPropertyAccess(target); - - requestProperties.forEach(p -> wrappedTarget.setPropertyValue(p, wrappedSource.getPropertyValue(p))); - } - } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index c2f1a4a9..f2d2259c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -193,7 +193,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess final BatchReceivePolicy batchReceivePolicy = new BatchReceivePolicy.Builder() .maxNumMessages(pulsarContainerProperties.getMaxNumMessages()) .maxNumBytes(pulsarContainerProperties.getMaxNumBytes()) - .timeout(pulsarContainerProperties.getBatchTimeout(), TimeUnit.MILLISECONDS).build(); + .timeout(pulsarContainerProperties.getBatchTimeoutMillis(), TimeUnit.MILLISECONDS).build(); this.consumer = getPulsarConsumerFactory().createConsumer( (Schema) pulsarContainerProperties.getSchema(), batchReceivePolicy, propertiesToConsumer); Assert.state(this.consumer != null, "Unable to create a consumer"); 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 fe91fcc1..b41f5966 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 @@ -58,7 +58,7 @@ public class PulsarContainerProperties { private int maxNumBytes = 10 * 1024 * 1024; - private int batchTimeout = 100; + private int batchTimeoutMillis = 100; private boolean batchListener; @@ -116,12 +116,12 @@ public class PulsarContainerProperties { this.maxNumBytes = maxNumBytes; } - public int getBatchTimeout() { - return this.batchTimeout; + public int getBatchTimeoutMillis() { + return this.batchTimeoutMillis; } - public void setBatchTimeout(int batchTimeout) { - this.batchTimeout = batchTimeout; + public void setBatchTimeoutMillis(int batchTimeoutMillis) { + this.batchTimeoutMillis = batchTimeoutMillis; } public boolean isBatchListener() { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java index 55a80331..5566f8a1 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java @@ -271,7 +271,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); - pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); CountDownLatch latch = new CountDownLatch(1); final PulsarBatchMessageListener pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class); @@ -319,7 +319,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests { PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); - pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); final PulsarBatchMessageListener pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class); CountDownLatch latch = new CountDownLatch(1); 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 c5915d4c..1627da6b 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 @@ -57,7 +57,7 @@ public class ConcurrentPulsarMessageListenerContainerTests { factory.setPulsarConsumerFactory(pulsarConsumerFactory); PulsarContainerProperties containerProperties = factory.getContainerProperties(); - containerProperties.setBatchTimeout(60_000); + containerProperties.setBatchTimeoutMillis(60_000); containerProperties.setMaxNumMessages(120); containerProperties.setMaxNumBytes(32000); @@ -69,7 +69,7 @@ public class ConcurrentPulsarMessageListenerContainerTests { final ConcurrentPulsarMessageListenerContainer concurrentContainer = factory .createListenerContainer(pulsarListenerEndpoint); final PulsarContainerProperties pulsarContainerProperties = concurrentContainer.getContainerProperties(); - assertThat(pulsarContainerProperties.getBatchTimeout()).isEqualTo(60_000); + assertThat(pulsarContainerProperties.getBatchTimeoutMillis()).isEqualTo(60_000); assertThat(pulsarContainerProperties.getMaxNumMessages()).isEqualTo(120); assertThat(pulsarContainerProperties.getMaxNumBytes()).isEqualTo(32_000); } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java index 71d272ff..3aac3a73 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java @@ -223,7 +223,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); - pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( PulsarBatchAcknowledgingMessageListener.class); @@ -294,7 +294,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); - pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( PulsarBatchAcknowledgingMessageListener.class); @@ -363,7 +363,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); - pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( PulsarBatchAcknowledgingMessageListener.class); @@ -432,7 +432,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); - pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( PulsarBatchAcknowledgingMessageListener.class); @@ -500,7 +500,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); - pulsarContainerProperties.setBatchTimeout(60_000); + pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); final PulsarBatchAcknowledgingMessageListener pulsarBatchMessageListener = mock( PulsarBatchAcknowledgingMessageListener.class);