From ed3899fcb9aee8e6767cdbe8275055a47fd22c47 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 17 Nov 2023 16:39:21 -0600 Subject: [PATCH] Make subscriptionType nullable on @Reactive/PulsarListener (#489) See #488 --- .../modules/ROOT/pages/reference/pulsar.adoc | 10 +- .../ROOT/pages/reference/reactive-pulsar.adoc | 13 +- .../annotation/ReactivePulsarListener.java | 8 +- ...arListenerAnnotationBeanPostProcessor.java | 14 +- .../ReactivePulsarContainerProperties.java | 2 +- .../listener/ReactivePulsarListenerTests.java | 179 +++++++++++++++++- .../pulsar/annotation/PulsarListener.java | 10 +- ...arListenerAnnotationBeanPostProcessor.java | 10 +- ...DefaultPulsarMessageListenerContainer.java | 39 +++- .../listener/PulsarContainerProperties.java | 2 +- .../pulsar/listener/PulsarListenerTests.java | 116 +++++++++++- 11 files changed, 373 insertions(+), 30 deletions(-) diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc index 2d294c05..bbadfca5 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc @@ -149,11 +149,15 @@ NOTE: If you are not using the starter, you will need to configure and register When it comes to Pulsar consumers, we recommend that end-user applications use the `PulsarListener` annotation. To use `PulsarListener`, you need to use the `@EnablePulsar` annotation. When you use Spring Boot support, it automatically enables this annotation and configures all the components necessary for `PulsarListener`, such as the message listener infrastructure (which is responsible for creating the Pulsar consumer). -`PulsarMessageListenerContainer` uses a `PulsarConsumerFactory` to create and manage the Pulsar consumer. +`PulsarMessageListenerContainer` uses a `PulsarConsumerFactory` to create and manage the Pulsar consumer the underlying Pulsar consumer that it uses to consume messages. -Spring Boot auto-configuration also provides this consumer factory which you can further configure by specifying **most** of the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties. +Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties. +**Most** of the configured properties on the factory will be respected in the listener with the following **exceptions**: + +TIP: The `spring.pulsar.consumer.subscription.name` property is ignored and is instead generated when not specified on the annotation. + +TIP: The `spring.pulsar.consumer.subscription-type` property is ignored and is instead taken from the value on the annotation. However, you can set the `subscriptionType = {}` on the annotation to instead use the property value as the default. -NOTE: `spring.pulsar.consumer.subscription.name` is ignored and is instead generated when not specified on the annotation. Let us revisit the `PulsarListener` code snippet we saw in the quick-tour section: diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc index 878e3ae2..36ae2f5d 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc @@ -159,8 +159,8 @@ Mono listen(String message) { } ---- -In this most basic form, when the `subscriptionName` is not provided on the `@ReactivePulsarListener` annotation an auto-generated subscription name will be used. -Likewise, when the `topics` are not directly provided, a <> is used to determine the destination topic. +In this most basic form, when the `topics` are not directly provided, a <> is used to determine the destination topic. +Likewise, when the `subscriptionName` is not provided on the `@ReactivePulsarListener` annotation an auto-generated subscription name will be used. In the `ReactivePulsarListener` method shown earlier, we receive the data as `String`, but we do not specify any schema types. Internally, the framework relies on Pulsar's schema mechanism to convert the data to the required type. @@ -238,9 +238,14 @@ Flux> listen2(Flux extends Abstra }); if (annotatedMethods.isEmpty()) { this.nonAnnotatedClasses.add(bean.getClass()); - this.logger.trace(() -> "No @PulsarListener annotations found on bean type: " + bean.getClass()); + this.logger + .trace(() -> "No @ReactivePulsarListener annotations found on bean type: " + bean.getClass()); } else { // Non-empty set of methods @@ -236,7 +237,7 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra endpoint.setId(getEndpointId(reactivePulsarListener)); endpoint.setTopics(topics); endpoint.setTopicPattern(topicPattern); - endpoint.setSubscriptionType(reactivePulsarListener.subscriptionType()); + resolveSubscriptionType(endpoint, reactivePulsarListener); endpoint.setSchemaType(reactivePulsarListener.schemaType()); String concurrency = reactivePulsarListener.concurrency(); @@ -260,6 +261,15 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor extends Abstra resolveConsumerCustomizer(endpoint, reactivePulsarListener); } + private void resolveSubscriptionType(MethodReactivePulsarListenerEndpoint endpoint, + ReactivePulsarListener reactivePulsarListener) { + Assert.state(reactivePulsarListener.subscriptionType().length <= 1, + () -> "ReactivePulsarListener.subscriptionType must have 0 or 1 elements"); + if (reactivePulsarListener.subscriptionType().length == 1) { + endpoint.setSubscriptionType(reactivePulsarListener.subscriptionType()[0]); + } + } + private void resolveDeadLetterPolicy(MethodReactivePulsarListenerEndpoint endpoint, ReactivePulsarListener reactivePulsarListener) { Object deadLetterPolicy = resolveExpression(reactivePulsarListener.deadLetterPolicy()); diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java index 2b0b851a..e60252d7 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java @@ -43,7 +43,7 @@ public class ReactivePulsarContainerProperties { private String subscriptionName; - private SubscriptionType subscriptionType = SubscriptionType.Exclusive; + private SubscriptionType subscriptionType; private Schema schema; diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java index 854667bb..d904d9c4 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java @@ -20,8 +20,9 @@ import static org.assertj.core.api.Assertions.assertThat; import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -45,10 +46,13 @@ import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory; import org.apache.pulsar.reactive.client.api.MessageResult; +import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer; +import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec; import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -71,13 +75,19 @@ import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsa import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer; import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig; import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.User2; +import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig; +import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig; import org.springframework.pulsar.support.PulsarHeaders; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.util.ObjectUtils; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -122,9 +132,14 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { return new PulsarTemplate<>(pulsarProducerFactory); } + @SuppressWarnings("unchecked") @Bean - public ReactivePulsarConsumerFactory pulsarConsumerFactory(ReactivePulsarClient pulsarClient) { - return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, Collections.emptyList()); + public ConsumerTrackingReactivePulsarConsumerFactory pulsarConsumerFactory( + ReactivePulsarClient pulsarClient, + ObjectProvider> defaultConsumerCustomizersProvider) { + DefaultReactivePulsarConsumerFactory consumerFactory = new DefaultReactivePulsarConsumerFactory<>( + pulsarClient, defaultConsumerCustomizersProvider.orderedStream().toList()); + return new ConsumerTrackingReactivePulsarConsumerFactory<>(consumerFactory); } @Bean @@ -721,7 +736,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } @Nested - @ContextConfiguration(classes = ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig.class) + @ContextConfiguration(classes = PulsarListenerWithHeadersConfig.class) class PulsarHeadersTest { static CountDownLatch simpleListenerLatch = new CountDownLatch(1); @@ -894,4 +909,160 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } + @Nested + class SubscriptionTypeTests { + + @Nested + @ContextConfiguration(classes = WithDefaultTypeConfig.class) + class WithDefaultType { + + static final CountDownLatch latchTypeNotSet = new CountDownLatch(1); + + @Test + void whenTypeNotSetAnywhereThenFallbackTypeIsUsed( + @Autowired ConsumerTrackingReactivePulsarConsumerFactory consumerFactory) throws Exception { + assertThat(consumerFactory.topicNameToConsumerSpec).hasEntrySatisfying("rpl-typeNotSetAnywhere-topic", + (consumerSpec) -> assertThat(consumerSpec.getSubscriptionType()) + .isEqualTo(SubscriptionType.Exclusive)); + pulsarTemplate.send("rpl-typeNotSetAnywhere-topic", "hello-rpl-typeNotSetAnywhere"); + assertThat(latchTypeNotSet.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Configuration(proxyBeanMethods = false) + static class WithDefaultTypeConfig { + + @ReactivePulsarListener(topics = "rpl-typeNotSetAnywhere-topic", + subscriptionName = "rpl-typeNotSetAnywhere-sub", + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenWithoutTypeSetAnywhere(String ignored) { + latchTypeNotSet.countDown(); + return Mono.empty(); + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithSpecificTypesConfig.class) + class WithSpecificTypes { + + static final CountDownLatch latchTypeSetConsumerFactory = new CountDownLatch(1); + + static final CountDownLatch latchTypeSetAnnotation = new CountDownLatch(1); + + static final CountDownLatch latchWithCustomizer = new CountDownLatch(1); + + @Test + void whenTypeSetOnlyInConsumerFactoryThenConsumerFactoryTypeIsUsed( + @Autowired ConsumerTrackingReactivePulsarConsumerFactory consumerFactory) throws Exception { + assertThat(consumerFactory.getSpec("rpl-typeSetConsumerFactory-topic")) + .extracting(ReactiveMessageConsumerSpec::getSubscriptionType) + .isEqualTo(SubscriptionType.Shared); + pulsarTemplate.send("rpl-typeSetConsumerFactory-topic", "hello-rpl-typeSetConsumerFactory"); + assertThat(latchTypeSetConsumerFactory.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void whenTypeSetOnAnnotationThenAnnotationTypeIsUsed( + @Autowired ConsumerTrackingReactivePulsarConsumerFactory consumerFactory) throws Exception { + assertThat(consumerFactory.getSpec("rpl-typeSetAnnotation-topic")) + .extracting(ReactiveMessageConsumerSpec::getSubscriptionType) + .isEqualTo(SubscriptionType.Key_Shared); + pulsarTemplate.send("rpl-typeSetAnnotation-topic", "hello-rpl-typeSetAnnotation"); + assertThat(latchTypeSetAnnotation.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void whenTypeSetWithCustomizerThenCustomizerTypeIsUsed( + @Autowired ConsumerTrackingReactivePulsarConsumerFactory consumerFactory) throws Exception { + assertThat(consumerFactory.getSpec("rpl-typeSetCustomizer-topic")) + .extracting(ReactiveMessageConsumerSpec::getSubscriptionType) + .isEqualTo(SubscriptionType.Failover); + pulsarTemplate.send("rpl-typeSetCustomizer-topic", "hello-rpl-typeSetCustomizer"); + assertThat(latchWithCustomizer.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Configuration(proxyBeanMethods = false) + static class WithSpecificTypesConfig { + + @Bean + ReactiveMessageConsumerBuilderCustomizer consumerFactoryDefaultSubTypeCustomizer() { + return (b) -> b.subscriptionType(SubscriptionType.Shared); + } + + @ReactivePulsarListener(topics = "rpl-typeSetConsumerFactory-topic", + subscriptionName = "rpl-typeSetConsumerFactory-sub", subscriptionType = {}, + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenWithTypeSetOnlyOnConsumerFactory(String ignored) { + latchTypeSetConsumerFactory.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(topics = "rpl-typeSetAnnotation-topic", + subscriptionName = "rpl-typeSetAnnotation-sub", subscriptionType = SubscriptionType.Key_Shared, + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenWithTypeSetOnAnnotation(String ignored) { + latchTypeSetAnnotation.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(topics = "rpl-typeSetCustomizer-topic", + subscriptionName = "rpl-typeSetCustomizer-sub", subscriptionType = SubscriptionType.Key_Shared, + consumerCustomizer = "myCustomizer") + Mono listenWithTypeSetInCustomizer(String ignored) { + latchWithCustomizer.countDown(); + return Mono.empty(); + } + + @Bean + public ReactivePulsarListenerMessageConsumerBuilderCustomizer myCustomizer() { + return cb -> cb.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscriptionType(SubscriptionType.Failover); + } + + } + + } + + } + + static class ConsumerTrackingReactivePulsarConsumerFactory implements ReactivePulsarConsumerFactory { + + private Map topicNameToConsumerSpec = new HashMap<>(); + + private ReactivePulsarConsumerFactory delegate; + + ConsumerTrackingReactivePulsarConsumerFactory(ReactivePulsarConsumerFactory delegate) { + this.delegate = delegate; + } + + @Override + public ReactiveMessageConsumer createConsumer(Schema schema) { + var consumer = this.delegate.createConsumer(schema); + storeSpec(consumer); + return consumer; + } + + @Override + public ReactiveMessageConsumer createConsumer(Schema schema, + List> reactiveMessageConsumerBuilderCustomizers) { + var consumer = this.delegate.createConsumer(schema, reactiveMessageConsumerBuilderCustomizers); + storeSpec(consumer); + return consumer; + } + + private void storeSpec(ReactiveMessageConsumer consumer) { + var consumerSpec = (ReactiveMessageConsumerSpec) ReflectionTestUtils.getField(consumer, "consumerSpec"); + var topicNamesKey = !ObjectUtils.isEmpty(consumerSpec.getTopicNames()) ? consumerSpec.getTopicNames().get(0) + : "no-topics-set"; + this.topicNameToConsumerSpec.put(topicNamesKey, consumerSpec); + } + + ReactiveMessageConsumerSpec getSpec(String topic) { + return this.topicNameToConsumerSpec.get(topic); + } + + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java index 8435cd65..5da37613 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java @@ -74,10 +74,12 @@ public @interface PulsarListener { String subscriptionName() default ""; /** - * Pulsar subscription type for this listener. - * @return the {@code subscriptionType} for this listener + * Pulsar subscription type for this listener - expected to be a single element array + * with subscription type or empty array to indicate null type. + * @return single element array with the subscription type or empty array to indicate + * no type chosen by user */ - SubscriptionType subscriptionType() default SubscriptionType.Exclusive; + SubscriptionType[] subscriptionType() default { SubscriptionType.Exclusive }; /** * Pulsar schema type for this listener. @@ -114,7 +116,7 @@ public @interface PulsarListener { * a {@link String}, in which case the {@link Boolean#parseBoolean(String)} is used to * obtain the value. *

- * SpEL {@code #{...}} and property place holders {@code ${...}} are supported. + * SpEL {@code #{...}} and property placeholders {@code ${...}} are supported. * @return true to auto start, false to not auto start. */ String autoStartup() default ""; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java index b271b1af..cc96bba0 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java @@ -226,7 +226,7 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar endpoint.setId(getEndpointId(pulsarListener)); endpoint.setTopics(topics); endpoint.setTopicPattern(topicPattern); - endpoint.setSubscriptionType(pulsarListener.subscriptionType()); + resolveSubscriptionType(endpoint, pulsarListener); endpoint.setSchemaType(pulsarListener.schemaType()); endpoint.setAckMode(pulsarListener.ackMode()); @@ -250,6 +250,14 @@ public class PulsarListenerAnnotationBeanPostProcessor extends AbstractPulsar resolveConsumerCustomizer(endpoint, pulsarListener); } + private void resolveSubscriptionType(MethodPulsarListenerEndpoint endpoint, PulsarListener pulsarListener) { + Assert.state(pulsarListener.subscriptionType().length <= 1, + () -> "PulsarListener.subscriptionType must have 0 or 1 elements"); + if (pulsarListener.subscriptionType().length == 1) { + endpoint.setSubscriptionType(pulsarListener.subscriptionType()[0]); + } + } + @SuppressWarnings({ "rawtypes" }) private void resolvePulsarConsumerErrorHandler(MethodPulsarListenerEndpoint endpoint, PulsarListener pulsarListener) { 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 d5d63bf1..3c599a82 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 @@ -46,6 +46,8 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.impl.ConsumerImpl; +import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.log.LogAccessor; @@ -62,6 +64,7 @@ import org.springframework.pulsar.observation.PulsarListenerObservation; import org.springframework.pulsar.observation.PulsarMessageReceiverContext; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; import io.micrometer.observation.Observation; @@ -226,7 +229,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess private final AckMode ackMode; - private final SubscriptionType subscriptionType; + private SubscriptionType subscriptionType; @SuppressWarnings({ "unchecked", "rawtypes" }) Listener(MessageListener messageListener, PulsarContainerProperties containerProperties) { @@ -280,12 +283,34 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess this.consumer = getPulsarConsumerFactory().createConsumer((Schema) containerProperties.getSchema(), topicNames, this.containerProperties.getSubscriptionName(), properties, customizers); Assert.state(this.consumer != null, "Unable to create a consumer"); + + // If our subscriptionType is null - update it based on the actual + // subscriptionType + // of the underlying consumer + if (this.subscriptionType == null) { + updateSubscriptionTypeFromConsumer(this.consumer); + } } catch (PulsarClientException e) { DefaultPulsarMessageListenerContainer.this.logger.error(e, () -> "Pulsar client exceptions."); } } + private void updateSubscriptionTypeFromConsumer(Consumer consumer) { + try { + var confField = ReflectionUtils.findField(ConsumerImpl.class, "conf"); + ReflectionUtils.makeAccessible(confField); + var conf = ReflectionUtils.getField(confField, consumer); + if (conf instanceof ConsumerConfigurationData confData) { + this.subscriptionType = confData.getSubscriptionType(); + } + } + catch (Exception ex) { + DefaultPulsarMessageListenerContainer.this.logger.error(ex, + () -> "Unable to determine default subscription type from consumer due to: " + ex.getMessage()); + } + } + private Map extractDirectConsumerProperties() { Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties(); return propertyOverrides.entrySet() @@ -410,6 +435,8 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } } catch (PulsarClientException pce) { + DefaultPulsarMessageListenerContainer.this.logger.warn(pce, + () -> "Batch acknowledgment failed: " + pce.getMessage()); this.consumer.negativeAcknowledge(messages); } } @@ -440,7 +467,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } // All the records are processed at this point. Handle acks. if (this.ackMode.equals(AckMode.BATCH)) { - handleAcks(messages); + handleBatchAcks(messages); } } } @@ -609,11 +636,11 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } private boolean isSharedSubscriptionType() { - return this.subscriptionType.equals(SubscriptionType.Shared) - || this.subscriptionType.equals(SubscriptionType.Key_Shared); + return this.subscriptionType != null && (this.subscriptionType.equals(SubscriptionType.Shared) + || this.subscriptionType.equals(SubscriptionType.Key_Shared)); } - private void handleAcks(Messages messages) { + private void handleBatchAcks(Messages messages) { if (this.nackableMessages.isEmpty()) { try { if (messages.size() > 0) { @@ -628,6 +655,8 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } } catch (PulsarClientException pce) { + DefaultPulsarMessageListenerContainer.this.logger.warn(pce, + () -> "Batch acknowledgments failed: " + pce.getMessage()); this.consumer.negativeAcknowledge(messages); } } 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 cc810ec1..8bfb5ee9 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 String subscriptionName; - private SubscriptionType subscriptionType = SubscriptionType.Exclusive; + private SubscriptionType subscriptionType; private Schema schema; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 80cb9c12..d63fe83e 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -40,16 +40,20 @@ import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; +import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.client.impl.schema.ProtobufSchema; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; +import org.assertj.core.api.AbstractObjectAssert; +import org.assertj.core.api.InstanceOfAssertFactories; import org.awaitility.Awaitility; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; @@ -61,6 +65,7 @@ import org.springframework.pulsar.annotation.PulsarListenerConsumerBuilderCustom import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; +import org.springframework.pulsar.core.ConsumerBuilderCustomizer; import org.springframework.pulsar.core.DefaultPulsarClientFactory; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; @@ -73,6 +78,8 @@ import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTopic; import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.core.TopicResolver; +import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig; +import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig; import org.springframework.pulsar.support.PulsarHeaders; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; import org.springframework.test.annotation.DirtiesContext; @@ -115,8 +122,10 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { } @Bean - public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, null); + public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, + ObjectProvider> defaultConsumerCustomizersProvider) { + return new DefaultPulsarConsumerFactory<>(pulsarClient, + defaultConsumerCustomizersProvider.orderedStream().toList()); } @Bean @@ -1103,4 +1112,107 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { } + @Nested + class SubscriptionTypeTests { + + @SuppressWarnings("rawtypes") + private static AbstractObjectAssert assertSubscriptionType(Consumer consumer) { + return assertThat(consumer) + .extracting("conf", InstanceOfAssertFactories.type(ConsumerConfigurationData.class)) + .extracting(ConsumerConfigurationData::getSubscriptionType); + } + + @Nested + @ContextConfiguration(classes = WithDefaultTypeConfig.class) + class WithDefaultType { + + static final CountDownLatch latchTypeNotSet = new CountDownLatch(1); + + @Test + void whenTypeNotSetAnywhereThenFallbackTypeIsUsed() throws Exception { + pulsarTemplate.send("typeNotSetAnywhere-topic", "hello-typeNotSetAnywhere"); + assertThat(latchTypeNotSet.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Configuration(proxyBeanMethods = false) + static class WithDefaultTypeConfig { + + @PulsarListener(topics = "typeNotSetAnywhere-topic", subscriptionName = "typeNotSetAnywhere-sub") + void listenWithoutTypeSetAnywhere(String ignored, Consumer consumer) { + assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Exclusive); + latchTypeNotSet.countDown(); + } + + } + + } + + @Nested + @ContextConfiguration(classes = WithSpecificTypesConfig.class) + class WithSpecificTypes { + + static final CountDownLatch latchTypeSetConsumerFactory = new CountDownLatch(1); + + static final CountDownLatch latchTypeSetAnnotation = new CountDownLatch(1); + + static final CountDownLatch latchWithCustomizer = new CountDownLatch(1); + + @Test + void whenTypeSetOnlyInConsumerFactoryThenConsumerFactoryTypeIsUsed() throws Exception { + pulsarTemplate.send("typeSetConsumerFactory-topic", "hello-typeSetConsumerFactory"); + assertThat(latchTypeSetConsumerFactory.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void whenTypeSetOnAnnotationThenAnnotationTypeIsUsed() throws Exception { + pulsarTemplate.send("typeSetAnnotation-topic", "hello-typeSetAnnotation"); + assertThat(latchTypeSetAnnotation.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void whenTypeSetWithCustomizerThenCustomizerTypeIsUsed() throws Exception { + pulsarTemplate.send("typeSetCustomizer-topic", "hello-typeSetCustomizer"); + assertThat(latchWithCustomizer.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Configuration(proxyBeanMethods = false) + static class WithSpecificTypesConfig { + + @Bean + ConsumerBuilderCustomizer consumerFactoryDefaultSubTypeCustomizer() { + return (b) -> b.subscriptionType(SubscriptionType.Shared); + } + + @PulsarListener(topics = "typeSetConsumerFactory-topic", + subscriptionName = "typeSetConsumerFactory-sub", subscriptionType = {}) + void listenWithTypeSetOnlyOnConsumerFactory(String ignored, Consumer consumer) { + assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Shared); + latchTypeSetConsumerFactory.countDown(); + } + + @PulsarListener(topics = "typeSetAnnotation-topic", subscriptionName = "typeSetAnnotation-sub", + subscriptionType = SubscriptionType.Key_Shared) + void listenWithTypeSetOnAnnotation(String ignored, Consumer consumer) { + assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Key_Shared); + latchTypeSetAnnotation.countDown(); + } + + @PulsarListener(topics = "typeSetCustomizer-topic", subscriptionName = "typeSetCustomizer-sub", + subscriptionType = SubscriptionType.Key_Shared, consumerCustomizer = "myCustomizer") + void listenWithTypeSetInCustomizer(String ignored, Consumer consumer) { + assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Failover); + latchWithCustomizer.countDown(); + } + + @Bean + public PulsarListenerConsumerBuilderCustomizer myCustomizer() { + return cb -> cb.subscriptionType(SubscriptionType.Failover); + } + + } + + } + + } + }