From fcdaf2fdd8d6c6a7e6fdd11077ed316268242e9b Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Thu, 14 Mar 2024 14:47:21 -0500 Subject: [PATCH] Fix Pulsar binder for Java 21 (#2918) Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2915 Running the Pulsar binder with Java 21 surfaced a bug where the layered binder/binding config props are not properly applied to the targeted consumer/producer builders. The bug existed before Java 21 and is unrelated. The binder/binding config props are converted to a Pulsar ProducerConfigurationData object via Jackson. This conversion process iterates over a map of config props. The iteration order of the map changed between Java 17 and 21. This results in object setters getting called in different orders, which in turn causes cross-field validation errors. See #2915 for more details --- .../binder/pulsar/PulsarBinderUtils.java | 116 ++++-- .../pulsar/PulsarMessageChannelBinder.java | 3 +- .../properties/ConsumerConfigProperties.java | 93 ++--- .../properties/ProducerConfigProperties.java | 16 +- .../pulsar/ConsumerConfigPropertiesTests.java | 91 +++-- .../pulsar/PulsarBinderIntegrationTests.java | 4 +- .../binder/pulsar/PulsarBinderUtilsTests.java | 333 ++++++++++++------ 7 files changed, 432 insertions(+), 224 deletions(-) diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java index 96cf8102d..f4600dfc8 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-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. @@ -16,11 +16,17 @@ package org.springframework.cloud.stream.binder.pulsar; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.UUID; +import org.apache.pulsar.client.api.BatcherBuilder; +import org.apache.pulsar.client.api.CryptoKeyReader; +import org.apache.pulsar.client.api.MessageRouter; +import org.apache.pulsar.client.api.ProducerBuilder; + import org.springframework.cloud.stream.binder.pulsar.properties.ConsumerConfigProperties; import org.springframework.cloud.stream.binder.pulsar.properties.ProducerConfigProperties; import org.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties; @@ -60,29 +66,32 @@ final class PulsarBinderUtils { /** * Merges base and extended producer properties defined at the binder and binding - * level. Only properties whose value has changed from the default are considered. If - * a property is defined at both the binder and binding level, the binding level - * property value is given precedence. + * level. + *

Only base properties whose value has changed from the default are included in + * result map. All extended properties are included, regardless of their value, + * which ensures that the extended property defaults are respected. + *

If a property is defined at both the binder and binding level, the binding level + * property value takes precedence. * @param binderProducerProps the binder level producer config properties (eg. * 'spring.cloud.stream.pulsar.binder.producer.*') * @param bindingProducerProps the binding level config properties (eg. * 'spring.cloud.stream.pulsar.bindings.myBinding-out-0.producer.*') - * @return map of modified merged binder and binding producer properties + * @return map of merged binder and binding producer properties */ static Map mergeModifiedProducerProperties(ProducerConfigProperties binderProducerProps, ProducerConfigProperties bindingProducerProps) { - // Layer the base props for common -> binder -> bindings - var baseProducerProps = new ProducerConfigProperties().toBaseProducerPropertiesMap(); + // Layer the base props for global -> binder -> bindings + var globalProducerProps = new ProducerConfigProperties().toBaseProducerPropertiesMap(); var binderBaseProducerProps = binderProducerProps.toBaseProducerPropertiesMap(); var bindingBaseProducerProps = bindingProducerProps.toBaseProducerPropertiesMap(); - var layeredBaseProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseProducerProps, - binderBaseProducerProps, bindingBaseProducerProps); - // Layer the extended props for binder -> bindings - var extProducerProps = new ProducerConfigProperties().toExtendedProducerPropertiesMap(); + var layeredBaseProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(globalProducerProps, + binderBaseProducerProps, bindingBaseProducerProps, false); + // Layer the extended props for global -> binder -> bindings + var globalExtProducerProps = new ProducerConfigProperties().toExtendedProducerPropertiesMap(); var binderExtProducerProps = binderProducerProps.toExtendedProducerPropertiesMap(); var bindingExtProducerProps = bindingProducerProps.toExtendedProducerPropertiesMap(); - var layeredExtProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(extProducerProps, - binderExtProducerProps, bindingExtProducerProps); + var layeredExtProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(globalExtProducerProps, + binderExtProducerProps, bindingExtProducerProps, true); // Combine both base and extended layers var layeredProducerProps = new HashMap<>(layeredBaseProducerProps); layeredProducerProps.putAll(layeredExtProducerProps); @@ -91,29 +100,32 @@ final class PulsarBinderUtils { /** * Merges base and extended consumer properties defined at the binder and binding - * level. Only properties whose value has changed from the default are considered. If - * a property is defined at both the binder and binding level, the binding level - * property value is given precedence. + * level. + *

Only base properties whose value has changed from the default are included in + * result map. All extended properties are included, regardless of their value, + * which ensures that the extended property defaults are respected. + *

If a property is defined at both the binder and binding level, the binding level + * property value takes precedence. * @param binderConsumerProps the binder level consumer config properties (eg. * 'spring.cloud.stream.pulsar.binder.consumer.*') * @param bindingConsumerProps the binding level config properties (eg. * 'spring.cloud.stream.pulsar.bindings.myBinding-in-0.consumer.*') - * @return map of modified merged binder and binding consumer properties + * @return map of merged binder and binding consumer properties */ static Map mergeModifiedConsumerProperties(ConsumerConfigProperties binderConsumerProps, ConsumerConfigProperties bindingConsumerProps) { - // Layer the base props for common -> binder -> bindings - var baseConsumerProps = new ConsumerConfigProperties().toBaseConsumerPropertiesMap(); + // Layer the base props for global -> binder -> bindings + var globalBaseConsumerProps = new ConsumerConfigProperties().toBaseConsumerPropertiesMap(); var binderBaseConsumerProps = binderConsumerProps.toBaseConsumerPropertiesMap(); var bindingBaseConsumerProps = bindingConsumerProps.toBaseConsumerPropertiesMap(); - var layeredBaseConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseConsumerProps, - binderBaseConsumerProps, bindingBaseConsumerProps); - // Layer the extended props for binder -> bindings - var extConsumerProps = new ConsumerConfigProperties().toExtendedConsumerPropertiesMap(); + var layeredBaseConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(globalBaseConsumerProps, + binderBaseConsumerProps, bindingBaseConsumerProps, false); + // Layer the extended props for global -> binder -> bindings + var globalExtConsumerProps = new ConsumerConfigProperties().toExtendedConsumerPropertiesMap(); var binderExtConsumerProps = binderConsumerProps.toExtendedConsumerPropertiesMap(); var bindingExtConsumerProps = bindingConsumerProps.toExtendedConsumerPropertiesMap(); - var layeredExtConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(extConsumerProps, - binderExtConsumerProps, bindingExtConsumerProps); + var layeredExtConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(globalExtConsumerProps, + binderExtConsumerProps, bindingExtConsumerProps, true); // Combine both base and extended layers var layeredConsumerProps = new HashMap<>(layeredBaseConsumerProps); layeredConsumerProps.putAll(layeredExtConsumerProps); @@ -121,35 +133,37 @@ final class PulsarBinderUtils { } /** - * Merges properties defined at the binder and binding level (binding properties - * override binder properties). + * Merges properties defined at the binder and binding level with binding properties + * overriding binder properties. *

- * NOTE: Properties whose value is not different from the default value in the - * {@code baseProps} are not included in the merged result. - * @param baseProps the map of base level properties (eg. 'spring.pulsar.consumer.*') + * @param globalProps the map of global level properties (eg. 'spring.pulsar.consumer.*') * @param binderProps the map of binder level properties (eg. * 'spring.cloud.stream.pulsar.binder.consumer.*') * @param bindingProps the map of binding level properties (eg. * 'spring.cloud.stream.pulsar.bindings.myBinding-in-0.consumer.*') - * @return map of merged binder and binding properties including only properties whose - * value has changed from the same property in the base properties + * @param includeDefaults whether to also include unmodified properties with their default values + * @return map of merged binder and binding properties with binding properties overriding binder properties */ - static Map mergePropertiesWithPrecedence(Map baseProps, - Map binderProps, Map bindingProps) { - Objects.requireNonNull(baseProps, "baseProps must be specified"); + static Map mergePropertiesWithPrecedence(Map globalProps, + Map binderProps, Map bindingProps, boolean includeDefaults) { + Objects.requireNonNull(globalProps, "globalProps must be specified"); Objects.requireNonNull(binderProps, "binderProps must be specified"); Objects.requireNonNull(bindingProps, "bindingProps must be specified"); - Map newOrModifiedBinderProps = extractNewOrModifiedProperties(binderProps, baseProps); + Map newOrModifiedBinderProps = extractNewOrModifiedProperties(binderProps, globalProps); LOGGER.trace(() -> "New or modified binder props: %s".formatted(newOrModifiedBinderProps)); - Map newOrModifiedBindingProps = extractNewOrModifiedProperties(bindingProps, baseProps); + Map newOrModifiedBindingProps = extractNewOrModifiedProperties(bindingProps, globalProps); LOGGER.trace(() -> "New or modified binding props: %s".formatted(newOrModifiedBindingProps)); Map mergedProps = new HashMap<>(newOrModifiedBinderProps); mergedProps.putAll(newOrModifiedBindingProps); - LOGGER.trace(() -> "Final merged props: %s".formatted(mergedProps)); + // Add in default properties for any props not customized by the user + if (includeDefaults) { + globalProps.forEach(mergedProps::putIfAbsent); + } + LOGGER.trace(() -> "Final merged props: %s".formatted(mergedProps)); return mergedProps; } @@ -164,4 +178,30 @@ final class PulsarBinderUtils { return newOrModifiedProps; } + /** + * Configures the specified properties onto the specified builder in a manner that + * loads non-serializable properties. See + * Pulsar PR. + * @param builder the builder + * @param properties the properties to set on the builder + * @param the payload type + */ + static void loadConf(ProducerBuilder builder, Map properties) { + builder.loadConf(properties); + // Set fields that are not loaded by loadConf + if (properties.containsKey("encryptionKeys")) { + @SuppressWarnings("unchecked") + Collection keys = (Collection) properties.get("encryptionKeys"); + keys.forEach(builder::addEncryptionKey); + } + if (properties.containsKey("customMessageRouter")) { + builder.messageRouter((MessageRouter) properties.get("customMessageRouter")); + } + if (properties.containsKey("batcherBuilder")) { + builder.batcherBuilder((BatcherBuilder) properties.get("batcherBuilder")); + } + if (properties.containsKey("cryptoKeyReader")) { + builder.cryptoKeyReader((CryptoKeyReader) properties.get("cryptoKeyReader")); + } + } } diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java index c933d5bd4..e571b8f37 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java @@ -46,7 +46,6 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.pulsar.core.ProducerBuilderConfigurationUtil; import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarTemplate; @@ -111,7 +110,7 @@ public class PulsarMessageChannelBinder extends var layeredProducerProps = PulsarBinderUtils.mergeModifiedProducerProperties( this.binderConfigProps.getProducer(), producerProperties.getExtension()); var handler = new PulsarProducerConfigurationMessageHandler(this.pulsarTemplate, schema, destination.getName(), - (builder) -> ProducerBuilderConfigurationUtil.loadConf(builder, layeredProducerProps), + (builder) -> PulsarBinderUtils.loadConf(builder, layeredProducerProps), determineOutboundHeaderMapper(producerProperties)); handler.setApplicationContext(getApplicationContext()); handler.setBeanFactory(getBeanFactory()); diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ConsumerConfigProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ConsumerConfigProperties.java index 7ba2db0c7..bbeb8f6da 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ConsumerConfigProperties.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ConsumerConfigProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-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. @@ -203,9 +203,31 @@ public class ConsumerConfigProperties extends PulsarProperties.Consumer { map.from(this::isRetryEnable).to(consumerProps.in("retryEnable")); map.from(this::getTopics).to(consumerProps.in("topicNames")); map.from(this::getTopicsPattern).to(consumerProps.in("topicsPattern")); + mapBaseSubscriptionProperties(this.getSubscription(), consumerProps, map); return consumerProps; } + private org.apache.pulsar.client.api.DeadLetterPolicy toPulsarDeadLetterPolicy(DeadLetterPolicy policy) { + Assert.state(policy.getMaxRedeliverCount() > 0, + "Pulsar DeadLetterPolicy must have a positive 'max-redelivery-count' property value"); + PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + org.apache.pulsar.client.api.DeadLetterPolicy.DeadLetterPolicyBuilder builder = org.apache.pulsar.client.api.DeadLetterPolicy + .builder(); + map.from(policy::getMaxRedeliverCount).to(builder::maxRedeliverCount); + map.from(policy::getRetryLetterTopic).to(builder::retryLetterTopic); + map.from(policy::getDeadLetterTopic).to(builder::deadLetterTopic); + map.from(policy::getInitialSubscriptionName).to(builder::initialSubscriptionName); + return builder.build(); + } + + private void mapBaseSubscriptionProperties(PulsarProperties.Consumer.Subscription subscription, Properties consumerProps, PropertyMapper map) { + map.from(subscription::getName).to(consumerProps.in("subscriptionName")); + map.from(subscription::getType).to(consumerProps.in("subscriptionType")); + map.from(subscription::getMode).to(consumerProps.in("subscriptionMode")); + map.from(subscription::getInitialPosition).to(consumerProps.in("subscriptionInitialPosition")); + map.from(subscription::getTopicsMode).to(consumerProps.in("regexSubscriptionMode")); + } + /** * Gets a map representation of the extended consumer properties (those defined in * this class). @@ -226,36 +248,38 @@ public class ConsumerConfigProperties extends PulsarProperties.Consumer { map.from(this::getReceiverQueueSize).to(consumerProps.in("receiverQueueSize")); map.from(this::getResetIncludeHead).to(consumerProps.in("resetIncludeHead")); map.from(this::getStartPaused).to(consumerProps.in("startPaused")); - // Acknowledgement properties - map.from(this::getAck).as(Acknowledgement::getGroupTime).as(it -> it.toNanos() / 1000) - .to(consumerProps.in("acknowledgementsGroupTimeMicros")); - map.from(this::getAck).as(Acknowledgement::getRedeliveryDelay).as(it -> it.toNanos() / 1000) - .to(consumerProps.in("negativeAckRedeliveryDelayMicros")); - map.from(this::getAck).as(Acknowledgement::getTimeout).as(Duration::toMillis) - .to(consumerProps.in("ackTimeoutMillis")); - map.from(this::getAck).as(Acknowledgement::getTimeoutTickDuration).as(Duration::toMillis) - .to(consumerProps.in("tickDurationMillis")); - map.from(this::getAck).as(Acknowledgement::getBatchIndexEnabled).to(consumerProps.in("batchIndexAckEnabled")); - map.from(this::getAck).as(Acknowledgement::getReceiptEnabled).to(consumerProps.in("ackReceiptEnabled")); - // Chunking properties - map.from(this::getChunk).as(Chunking::getExpireTimeIncomplete).as(Duration::toMillis) - .to(consumerProps.in("expireTimeOfIncompleteChunkedMessageMillis")); - map.from(this::getChunk).as(Chunking::getAutoAckOldestOnQueueFull) - .to(consumerProps.in("autoAckOldestChunkedMessageOnQueueFull")); - map.from(this::getChunk).as(Chunking::getMaxPendingMessages).to(consumerProps.in("maxPendingChunkedMessage")); - // Subscription properties - map.from(this::getSubscription).as(Subscription::getName).to(consumerProps.in("subscriptionName")); - map.from(this::getSubscription).as(Subscription::getType).to(consumerProps.in("subscriptionType")); - map.from(this::getSubscription).as(Subscription::getProperties).to(consumerProps.in("subscriptionProperties")); - map.from(this::getSubscription).as(Subscription::getMode).to(consumerProps.in("subscriptionMode")); - map.from(this::getSubscription).as(Subscription::getInitialPosition) - .to(consumerProps.in("subscriptionInitialPosition")); - map.from(this::getSubscription).as(Subscription::getTopicsMode).to(consumerProps.in("regexSubscriptionMode")); - map.from(this::getSubscription).as(Subscription::getReplicateState) - .to(consumerProps.in("replicateSubscriptionState")); + this.mapAcknowledgementProperties(this.getAck(), consumerProps, map); + this.mapChunkProperties(this.getChunk(), consumerProps, map); + this.mapExtendedSubscriptionProperties(this.getSubscription(), consumerProps, map); return consumerProps; } + private void mapAcknowledgementProperties(Acknowledgement ack, Properties consumerProps, PropertyMapper map) { + map.from(ack::getGroupTime).as(it -> it.toNanos() / 1000) + .to(consumerProps.in("acknowledgementsGroupTimeMicros")); + map.from(ack::getRedeliveryDelay).as(it -> it.toNanos() / 1000) + .to(consumerProps.in("negativeAckRedeliveryDelayMicros")); + map.from(ack::getTimeout).as(Duration::toMillis) + .to(consumerProps.in("ackTimeoutMillis")); + map.from(ack::getTimeoutTickDuration).as(Duration::toMillis) + .to(consumerProps.in("tickDurationMillis")); + map.from(ack::getBatchIndexEnabled).to(consumerProps.in("batchIndexAckEnabled")); + map.from(ack::getReceiptEnabled).to(consumerProps.in("ackReceiptEnabled")); + } + + private void mapChunkProperties(Chunking chunk, Properties consumerProps, PropertyMapper map) { + map.from(chunk::getExpireTimeIncomplete).as(Duration::toMillis) + .to(consumerProps.in("expireTimeOfIncompleteChunkedMessageMillis")); + map.from(chunk::getAutoAckOldestOnQueueFull) + .to(consumerProps.in("autoAckOldestChunkedMessageOnQueueFull")); + map.from(chunk::getMaxPendingMessages).to(consumerProps.in("maxPendingChunkedMessage")); + } + + private void mapExtendedSubscriptionProperties(Subscription subscription, Properties consumerProps, PropertyMapper map) { + map.from(subscription::getProperties).to(consumerProps.in("subscriptionProperties")); + map.from(subscription::getReplicateState).to(consumerProps.in("replicateSubscriptionState")); + } + /** * Gets a map representation of base and extended consumer properties. * @return map of base and extended consumer properties and associated values. @@ -266,19 +290,6 @@ public class ConsumerConfigProperties extends PulsarProperties.Consumer { return consumerProps; } - private org.apache.pulsar.client.api.DeadLetterPolicy toPulsarDeadLetterPolicy(DeadLetterPolicy policy) { - Assert.state(policy.getMaxRedeliverCount() > 0, - "Pulsar DeadLetterPolicy must have a positive 'max-redelivery-count' property value"); - PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); - org.apache.pulsar.client.api.DeadLetterPolicy.DeadLetterPolicyBuilder builder = org.apache.pulsar.client.api.DeadLetterPolicy - .builder(); - map.from(policy::getMaxRedeliverCount).to(builder::maxRedeliverCount); - map.from(policy::getRetryLetterTopic).to(builder::retryLetterTopic); - map.from(policy::getDeadLetterTopic).to(builder::deadLetterTopic); - map.from(policy::getInitialSubscriptionName).to(builder::initialSubscriptionName); - return builder.build(); - } - public static class Acknowledgement { /** diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ProducerConfigProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ProducerConfigProperties.java index 64188f08a..a3f871ba3 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ProducerConfigProperties.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ProducerConfigProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-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. @@ -230,16 +230,20 @@ public class ProducerConfigProperties extends PulsarProperties.Producer { .to(producerProps.in("maxPendingMessagesAcrossPartitions")); map.from(this::getMultiSchema).to(producerProps.in("multiSchema")); map.from(this::getProperties).to(producerProps.in("properties")); + this.mapBatchProperties(this.getBatch(), producerProps, map); + return producerProps; + } + + private void mapBatchProperties(Batching batch, Properties producerProps, PropertyMapper map) { if (this.isBatchingEnabled()) { - map.from(this::getBatch).as(Batching::getMaxPublishDelay).as(it -> it.toNanos() / 1000) + map.from(batch::getMaxPublishDelay).as(it -> it.toNanos() / 1000) .to(producerProps.in("batchingMaxPublishDelayMicros")); - map.from(this::getBatch).as(Batching::getPartitionSwitchFrequencyByPublishDelay) + map.from(batch::getPartitionSwitchFrequencyByPublishDelay) .to(producerProps.in("batchingPartitionSwitchFrequencyByPublishDelay")); - map.from(this::getBatch).as(Batching::getMaxMessages).to(producerProps.in("batchingMaxMessages")); - map.from(this::getBatch).as(Batching::getMaxBytes).asInt(DataSize::toBytes) + map.from(batch::getMaxMessages).to(producerProps.in("batchingMaxMessages")); + map.from(batch::getMaxBytes).asInt(DataSize::toBytes) .to(producerProps.in("batchingMaxBytes")); } - return producerProps; } /** diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/ConsumerConfigPropertiesTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/ConsumerConfigPropertiesTests.java index 8a5484de1..ce5caad97 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/ConsumerConfigPropertiesTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/ConsumerConfigPropertiesTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-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. @@ -37,6 +37,7 @@ import org.springframework.cloud.stream.binder.pulsar.properties.ConsumerConfigP import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatNoException; +import static org.assertj.core.api.Assertions.entry; /** * Unit tests for {@link ConsumerConfigProperties}. @@ -47,7 +48,7 @@ import static org.assertj.core.api.Assertions.assertThatNoException; class ConsumerConfigPropertiesTests { @Test - void basePropsCanBeExtractedToMap() { + void allBasePropsCanBeExtractedToMap() { var inputProps = basePropsInputMap(); var consumerConfigProps = bindInputPropsToConsumerConfigProps(inputProps); var outputProps = consumerConfigProps.toBaseConsumerPropertiesMap(); @@ -55,35 +56,60 @@ class ConsumerConfigPropertiesTests { verifyBasePropsInOutputMap(outputProps); } + @Test + void nullBasePropsSkippedWhenExtractedToMap() { + var inputProps = Map.of("spring.pulsar.consumer.name", "my-consumer"); + var consumerConfigProps = bindInputPropsToConsumerConfigProps(inputProps); + var outputProps = consumerConfigProps.toBaseConsumerPropertiesMap(); + assertThat(outputProps).contains(entry("consumerName", "my-consumer")); + assertThat(outputProps).doesNotContainKey("deadLetterPolicy"); + } + private Map basePropsInputMap() { Map inputProps = new HashMap<>(); - inputProps.put("spring.pulsar.consumer.dead-letter-policy.max-redeliver-count", "4"); - inputProps.put("spring.pulsar.consumer.dead-letter-policy.retry-letter-topic", "my-retry-topic"); - inputProps.put("spring.pulsar.consumer.dead-letter-policy.dead-letter-topic", "my-dlt-topic"); - inputProps.put("spring.pulsar.consumer.dead-letter-policy.initial-subscription-name", - "my-initial-subscription"); inputProps.put("spring.pulsar.consumer.name", "my-consumer"); inputProps.put("spring.pulsar.consumer.priority-level", "8"); inputProps.put("spring.pulsar.consumer.read-compacted", "true"); inputProps.put("spring.pulsar.consumer.retry-enable", "true"); inputProps.put("spring.pulsar.consumer.topics[0]", "my-topic"); inputProps.put("spring.pulsar.consumer.topics-pattern", "my-pattern"); + // DeadLetterPolicy + inputProps.put("spring.pulsar.consumer.dead-letter-policy.max-redeliver-count", "4"); + inputProps.put("spring.pulsar.consumer.dead-letter-policy.retry-letter-topic", "my-retry-topic"); + inputProps.put("spring.pulsar.consumer.dead-letter-policy.dead-letter-topic", "my-dlt-topic"); + inputProps.put("spring.pulsar.consumer.dead-letter-policy.initial-subscription-name", + "my-initial-subscription"); + // Subscription + inputProps.put("spring.pulsar.consumer.subscription.name", "my-subscription"); + inputProps.put("spring.pulsar.consumer.subscription.type", "exclusive"); + inputProps.put("spring.pulsar.consumer.subscription.mode", "non-durable"); + inputProps.put("spring.pulsar.consumer.subscription.initial-position", "earliest"); + inputProps.put("spring.pulsar.consumer.subscription.topics-mode", "all-topics"); return inputProps; } private void verifyBasePropsInOutputMap(Map outputProps) { - assertThat(outputProps).hasEntrySatisfying("deadLetterPolicy", dlp -> { - DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) dlp; - assertThat(deadLetterPolicy.getMaxRedeliverCount()).isEqualTo(4); - assertThat(deadLetterPolicy.getRetryLetterTopic()).isEqualTo("my-retry-topic"); - assertThat(deadLetterPolicy.getDeadLetterTopic()).isEqualTo("my-dlt-topic"); - assertThat(deadLetterPolicy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription"); - }).containsEntry("consumerName", "my-consumer").containsEntry("priorityLevel", 8) - .containsEntry("readCompacted", true).containsEntry("retryEnable", true) - .hasEntrySatisfying("topicNames", - topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.collection(String.class)) + assertThat(outputProps) + .containsEntry("consumerName", "my-consumer") + .containsEntry("priorityLevel", 8) + .containsEntry("readCompacted", true) + .containsEntry("retryEnable", true) + .hasEntrySatisfying("topicNames", topics -> + assertThat(topics).asInstanceOf(InstanceOfAssertFactories.collection(String.class)) .containsExactly("my-topic")) - .hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern")); + .hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern")) + .containsEntry("subscriptionName", "my-subscription") + .containsEntry("subscriptionType", SubscriptionType.Exclusive) + .containsEntry("subscriptionMode", SubscriptionMode.NonDurable) + .containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest) + .containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics) + .hasEntrySatisfying("deadLetterPolicy", dlp -> { + DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) dlp; + assertThat(deadLetterPolicy.getMaxRedeliverCount()).isEqualTo(4); + assertThat(deadLetterPolicy.getRetryLetterTopic()).isEqualTo("my-retry-topic"); + assertThat(deadLetterPolicy.getDeadLetterTopic()).isEqualTo("my-dlt-topic"); + assertThat(deadLetterPolicy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription"); + }); } @Test @@ -119,21 +145,19 @@ class ConsumerConfigPropertiesTests { inputProps.put("spring.pulsar.consumer.receiver-queue-size", "1"); inputProps.put("spring.pulsar.consumer.reset-include-head", "true"); inputProps.put("spring.pulsar.consumer.start-paused", "true"); + // Acknowledgment inputProps.put("spring.pulsar.consumer.ack.group-time", "2s"); inputProps.put("spring.pulsar.consumer.ack.redelivery-delay", "3s"); inputProps.put("spring.pulsar.consumer.ack.timeout", "6s"); inputProps.put("spring.pulsar.consumer.ack.timeout-tick-duration", "7s"); inputProps.put("spring.pulsar.consumer.ack.batch-index-enabled", "true"); inputProps.put("spring.pulsar.consumer.ack.receipt-enabled", "true"); + // Chunk inputProps.put("spring.pulsar.consumer.chunk.expire-time-incomplete", "12s"); inputProps.put("spring.pulsar.consumer.chunk.auto-ack-oldest-on-queue-full", "false"); inputProps.put("spring.pulsar.consumer.chunk.max-pending-messages", "11"); - inputProps.put("spring.pulsar.consumer.subscription.name", "my-subscription"); - inputProps.put("spring.pulsar.consumer.subscription.type", "shared"); + // Subscription inputProps.put("spring.pulsar.consumer.subscription.properties[my-sub-prop]", "my-sub-prop-value"); - inputProps.put("spring.pulsar.consumer.subscription.mode", "nondurable"); - inputProps.put("spring.pulsar.consumer.subscription.initial-position", "earliest"); - inputProps.put("spring.pulsar.consumer.subscription.topics-mode", "all-topics"); inputProps.put("spring.pulsar.consumer.subscription.replicate-state", "true"); return inputProps; } @@ -143,27 +167,28 @@ class ConsumerConfigPropertiesTests { .containsEntry("autoUpdatePartitionsIntervalSeconds", 10L) .containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD) .containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5) - .containsEntry("patternAutoDiscoveryPeriod", 9).containsEntry("poolMessages", true) + .containsEntry("patternAutoDiscoveryPeriod", 9) + .containsEntry("poolMessages", true) .hasEntrySatisfying("properties", properties -> assertThat(properties) .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) .containsEntry("my-prop", "my-prop-value")) - .containsEntry("receiverQueueSize", 1).containsEntry("resetIncludeHead", true) - .containsEntry("startPaused", true).containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L) - .containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L).containsEntry("ackTimeoutMillis", 6_000L) - .containsEntry("tickDurationMillis", 7_000L).containsEntry("batchIndexAckEnabled", true) + .containsEntry("receiverQueueSize", 1) + .containsEntry("resetIncludeHead", true) + .containsEntry("startPaused", true) + .containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L) + .containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L) + .containsEntry("ackTimeoutMillis", 6_000L) + .containsEntry("tickDurationMillis", 7_000L) + .containsEntry("batchIndexAckEnabled", true) .containsEntry("ackReceiptEnabled", true) .containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L) .containsEntry("autoAckOldestChunkedMessageOnQueueFull", false) - .containsEntry("maxPendingChunkedMessage", 11).containsEntry("subscriptionName", "my-subscription") - .containsEntry("subscriptionType", SubscriptionType.Shared) + .containsEntry("maxPendingChunkedMessage", 11) .hasEntrySatisfying("subscriptionProperties", properties -> assertThat(properties) .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) .containsEntry("my-sub-prop", "my-sub-prop-value")) - .containsEntry("subscriptionMode", SubscriptionMode.NonDurable) - .containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest) - .containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics) .containsEntry("replicateSubscriptionState", true); } diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java index 90d3ac0a5..d02b1d201 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java @@ -88,7 +88,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { "--spring.pulsar.producer.name=textSupplierProducer-fromBase", "--spring.cloud.stream.pulsar.binder.producer.name=textSupplierProducer-fromBinder", "--spring.cloud.stream.pulsar.bindings.textSupplier-out-0.producer.name=textSupplierProducer-fromBinding", -// "--spring.cloud.stream.pulsar.binder.producer.max-pending-messages=1100", + "--spring.cloud.stream.pulsar.binder.producer.max-pending-messages=1100", "--spring.cloud.stream.pulsar.binder.producer.block-if-queue-full=true", "--spring.cloud.stream.pulsar.binder.consumer.subscription.name=textLoggerSub-fromBinder", "--spring.cloud.stream.pulsar.binder.consumer.name=textLogger-fromBinder", @@ -101,7 +101,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { TrackingProducerFactory producerFactory = context.getBean(TrackingProducerFactory.class); assertThat(producerFactory.producersCreated).isNotEmpty().element(0) .hasFieldOrPropertyWithValue("producerName", "textSupplierProducer-fromBinding") -// .hasFieldOrPropertyWithValue("conf.maxPendingMessages", 1100) + .hasFieldOrPropertyWithValue("conf.maxPendingMessages", 1100) .hasFieldOrPropertyWithValue("conf.blockIfQueueFull", true); TrackingConsumerFactory consumerFactory = context.getBean(TrackingConsumerFactory.class); diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java index 2c90cd56b..8eaf2fb1b 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-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. @@ -76,60 +76,99 @@ class PulsarBinderUtilsTests { @ParameterizedTest(name = "{0}") @MethodSource("mergePropertiesTestProvider") - void mergePropertiesTest(String testName, Map baseProps, Map binderProps, + void mergePropertiesTest(String testName, boolean includeDefaults, Map globalProps, Map binderProps, Map bindingProps, Map expectedMergedProps) { - assertThat(PulsarBinderUtils.mergePropertiesWithPrecedence(baseProps, binderProps, bindingProps)) + assertThat(PulsarBinderUtils.mergePropertiesWithPrecedence(globalProps, binderProps, bindingProps, includeDefaults)) .containsExactlyInAnyOrderEntriesOf(expectedMergedProps); } // @formatter:off static Stream mergePropertiesTestProvider() { return Stream.of( - arguments("binderLevelContainsSamePropAsBaseWithDiffValue", - Map.of("foo", "foo-base"), - Map.of("foo", "foo-binder"), + arguments("noProps", + true, Collections.emptyMap(), - Map.of("foo", "foo-binder")), - arguments("binderLevelContainsNewPropNotInBase", + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap()), + arguments("allSamePropsWithoutDefaults", + false, + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base"), + Collections.emptyMap()), + arguments("allSamePropsWithDefaults", + true, + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base")), + arguments("onlyBasePropsWithoutDefaults", + false, + Map.of("foo", "foo-base"), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap()), + arguments("onlyBasePropsWithDefaults", + true, + Map.of("foo", "foo-base"), + Collections.emptyMap(), + Collections.emptyMap(), + Map.of("foo", "foo-base")), + arguments("onlyBinderProps", + true, Collections.emptyMap(), Map.of("foo", "foo-binder"), Collections.emptyMap(), Map.of("foo", "foo-binder")), - arguments("binderLevelContainsSamePropAsBaseWithSameValue", - Map.of("foo", "foo-base"), - Map.of("foo", "foo-base"), + arguments("onlyBindingProps", + true, Collections.emptyMap(), - Collections.emptyMap()), - arguments("bindingLevelContainsSamePropAsBaseWithDiffValue", - Map.of("foo", "foo-base"), Collections.emptyMap(), Map.of("foo", "foo-binding"), Map.of("foo", "foo-binding")), - arguments("bindingLevelContainsNewPropNotInBase", - Collections.emptyMap(), - Map.of("foo", "foo-binding"), - Collections.emptyMap(), - Map.of("foo", "foo-binding")), - arguments("bindingLevelContainsSamePropAsBaseWithSameValue", + arguments("binderOverridesBaseValue", + true, Map.of("foo", "foo-base"), - Collections.emptyMap(), - Map.of("foo", "foo-base"), - Collections.emptyMap()), - arguments("bindingOverridesBinder", - Map.of("bar", "bar-base"), Map.of("foo", "foo-binder"), + Map.of("foo", "foo-base"), + Map.of("foo", "foo-binder")), + arguments("binderContainsNewPropWithoutDefaults", + false, + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base", "bar", "bar-binder"), + Map.of("foo", "foo-base"), + Map.of("bar", "bar-binder")), + arguments("binderContainsNewPropWithDefaults", + true, + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base", "bar", "bar-binder"), + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base", "bar", "bar-binder")), + arguments("bindingOverridesBaseValue", + true, + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base"), Map.of("foo", "foo-binding"), Map.of("foo", "foo-binding")), + arguments("bindingContainsNewPropWithoutDefaults", + false, + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base", "bar", "bar-binding"), + Map.of("bar", "bar-binding")), + arguments("bindingContainsNewPropWithDefaults", + true, + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base"), + Map.of("foo", "foo-base", "bar", "bar-binding"), + Map.of("foo", "foo-base", "bar", "bar-binding")), arguments("binderOverridesBaseAndBindingOverridesBinder", + true, Map.of("foo", "foo-base"), Map.of("foo", "foo-binder"), Map.of("foo", "foo-binding"), - Map.of("foo", "foo-binding")), - arguments("onlyBaseProps", - Map.of("foo", "foo-base"), - Collections.emptyMap(), - Collections.emptyMap(), - Collections.emptyMap())); + Map.of("foo", "foo-binding"))); } // @formatter:on @@ -142,77 +181,120 @@ class PulsarBinderUtilsTests { }; @Test - void noPropsSpecified() { - doMergeProducerPropertiesTest(SET_NO_PROPS, SET_NO_PROPS, Collections.emptyMap()); + void noPropsModified() { + var expectedProps = defaultExtProps(); + doMergeProducerPropertiesTest(SET_NO_PROPS, SET_NO_PROPS, expectedProps); + } + + // @formatter:off + @Test + void basePropModifiedAtBinderLevel() { + var expectedProps = defaultExtPropsWith("accessMode", ProducerAccessMode.Exclusive); + doMergeProducerPropertiesTest( + (binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Exclusive), + SET_NO_PROPS, + expectedProps); } @Test - void basePropSpecifiedAtBinderLevelOnly() { - doMergeProducerPropertiesTest((binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Exclusive), - SET_NO_PROPS, Map.of("accessMode", ProducerAccessMode.Exclusive)); - } - - @Test - void basePropSpecifiedAtBindingLevelOnly() { - doMergeProducerPropertiesTest(SET_NO_PROPS, + void basePropModifiedAtBindingLevel() { + var expectedProps = defaultExtPropsWith("accessMode", ProducerAccessMode.Exclusive); + doMergeProducerPropertiesTest( + SET_NO_PROPS, (bindingProps) -> bindingProps.setAccessMode(ProducerAccessMode.Exclusive), - Map.of("accessMode", ProducerAccessMode.Exclusive)); + expectedProps); } @Test - void basePropSpecifiedAtBinderAndBindingLevel() { + void basePropModifiedAtBinderAndBindingLevel() { + var expectedProps = defaultExtPropsWith("accessMode", ProducerAccessMode.Exclusive); doMergeProducerPropertiesTest( (binderProps) -> binderProps.setAccessMode(ProducerAccessMode.ExclusiveWithFencing), (bindingProps) -> bindingProps.setAccessMode(ProducerAccessMode.Exclusive), - Map.of("accessMode", ProducerAccessMode.Exclusive)); + expectedProps); } @Test - void basePropSpecifiedWithSameValueAsDefault() { - doMergeProducerPropertiesTest((binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Shared), - (bindingProps) -> bindingProps.setAccessMode(ProducerAccessMode.Shared), Collections.emptyMap()); + void basePropModifiedAtBinderAndBindingLevelWithDefaultValue() { + var expectedProps = defaultExtProps(); + doMergeProducerPropertiesTest( + (binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Shared), + (bindingProps) -> bindingProps.setAccessMode(ProducerAccessMode.Shared), + expectedProps); } @Test - void extPropSpecifiedAtBinderLevelOnly() { - doMergeProducerPropertiesTest((binderProps) -> binderProps.setMaxPendingMessages(1200), SET_NO_PROPS, - Map.of("maxPendingMessages", 1200)); + void extPropModifiedAtBinderLevel() { + var expectedProps = defaultExtPropsWith((p) -> p.setMaxPendingMessages(1200)); + doMergeProducerPropertiesTest( + (binderProps) -> binderProps.setMaxPendingMessages(1200), + SET_NO_PROPS, + expectedProps); } @Test - void extPropSpecifiedAtBindingLevelOnly() { - doMergeProducerPropertiesTest(SET_NO_PROPS, (bindingProps) -> bindingProps.setMaxPendingMessages(1200), - Map.of("maxPendingMessages", 1200)); + void extPropModifiedAtBindingLevel() { + var expectedProps = defaultExtPropsWith((p) -> p.setMaxPendingMessages(1200)); + doMergeProducerPropertiesTest( + SET_NO_PROPS, + (binderProps) -> binderProps.setMaxPendingMessages(1200), + expectedProps); } @Test - void extPropSpecifiedAtBinderAndBindingLevel() { - doMergeProducerPropertiesTest((binderProps) -> binderProps.setMaxPendingMessages(1100), - (bindingProps) -> bindingProps.setMaxPendingMessages(1200), Map.of("maxPendingMessages", 1200)); + void extPropModifiedAtBinderAndBindingLevel() { + var expectedProps = defaultExtPropsWith((p) -> p.setMaxPendingMessages(1200)); + doMergeProducerPropertiesTest( + (binderProps) -> binderProps.setMaxPendingMessages(1100), + (bindingProps) -> bindingProps.setMaxPendingMessages(1200), + expectedProps); } @Test - void extPropSpecifiedWithSameValueAsDefault() { - doMergeProducerPropertiesTest((binderProps) -> binderProps.setMaxPendingMessages(1000), - (bindingProps) -> bindingProps.setMaxPendingMessages(1000), Collections.emptyMap()); + void extPropModifiedAtBinderAndBindingLevelWithDefaultValue() { + var expectedProps = defaultExtProps(); + doMergeProducerPropertiesTest( + (binderProps) -> binderProps.setMaxPendingMessages(1000), + (bindingProps) -> bindingProps.setMaxPendingMessages(1000), + expectedProps); } @Test void baseAndExtPropsAreCombined() { - doMergeProducerPropertiesTest((binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Exclusive), + var expectedProps = defaultExtPropsWith((p) -> p.setMaxPendingMessages(1200)); + expectedProps.put("accessMode", ProducerAccessMode.Exclusive); + doMergeProducerPropertiesTest( + (binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Exclusive), (bindingProps) -> bindingProps.setMaxPendingMessages(1200), - Map.of("accessMode", ProducerAccessMode.Exclusive, "maxPendingMessages", 1200)); + expectedProps); + } + // @formatter:on + + private Map defaultExtProps() { + return new ProducerConfigProperties().toExtendedProducerPropertiesMap(); } - void doMergeProducerPropertiesTest(Consumer binderPropsCustomizer, - Consumer bindingPropsCustomizer, Map expectedProps) { + private Map defaultExtPropsWith(String key, Object value) { + var defaultExtProps = defaultExtProps(); + defaultExtProps.put(key, value); + return defaultExtProps; + } + + private Map defaultExtPropsWith(Consumer extPropsCustomizer) { + var extProps = new ProducerConfigProperties(); + extPropsCustomizer.accept(extProps); + return extProps.toExtendedProducerPropertiesMap(); + } + + private void doMergeProducerPropertiesTest(Consumer binderPropsCustomizer, + Consumer bindingPropsCustomizer, + Map expectedMergedProperties) { var binderProducerProps = new ProducerConfigProperties(); binderPropsCustomizer.accept(binderProducerProps); var bindingProducerProps = new ProducerConfigProperties(); bindingPropsCustomizer.accept(bindingProducerProps); - var mergedProps = PulsarBinderUtils.mergeModifiedProducerProperties(binderProducerProps, - bindingProducerProps); - assertThat(mergedProps).containsExactlyInAnyOrderEntriesOf(expectedProps); + var mergedProps = PulsarBinderUtils.mergeModifiedProducerProperties(binderProducerProps, bindingProducerProps); + assertThat(mergedProps).containsExactlyInAnyOrderEntriesOf(expectedMergedProperties); } } @@ -224,74 +306,121 @@ class PulsarBinderUtilsTests { }; @Test - void noPropsSpecified() { - doMergeConsumerPropertiesTest(SET_NO_PROPS, SET_NO_PROPS, Collections.emptyMap()); + void noPropsModified() { + var expectedProps = defaultExtProps(); + doMergeConsumerPropertiesTest(SET_NO_PROPS, SET_NO_PROPS, expectedProps); + } + + // @formatter:off + @Test + void basePropModifiedAtBinderLevel() { + var expectedProps = defaultExtPropsWith("priorityLevel", 1000); + doMergeConsumerPropertiesTest( + (binderProps) -> binderProps.setPriorityLevel(1000), + SET_NO_PROPS, + expectedProps); } @Test - void basePropSpecifiedAtBinderLevelOnly() { - doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(1000), SET_NO_PROPS, - Map.of("priorityLevel", 1000)); + void basePropModifiedAtBindingLevel() { + var expectedProps = defaultExtPropsWith("priorityLevel", 1000); + doMergeConsumerPropertiesTest( + SET_NO_PROPS, + (bindingProps) -> bindingProps.setPriorityLevel(1000), + expectedProps); } @Test - void basePropSpecifiedAtBindingLevelOnly() { - doMergeConsumerPropertiesTest(SET_NO_PROPS, (bindingProps) -> bindingProps.setPriorityLevel(1000), - Map.of("priorityLevel", 1000)); + void basePropModifiedAtBinderAndBindingLevel() { + var expectedProps = defaultExtPropsWith("priorityLevel", 1000); + doMergeConsumerPropertiesTest( + (binderProps) -> binderProps.setPriorityLevel(2000), + (bindingProps) -> bindingProps.setPriorityLevel(1000), + expectedProps); } @Test - void basePropSpecifiedAtBinderAndBindingLevel() { - doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(2000), - (bindingProps) -> bindingProps.setPriorityLevel(1000), Map.of("priorityLevel", 1000)); + void basePropModifiedAtBinderAndBindingLevelWithDefaultValue() { + var expectedProps = defaultExtProps(); + doMergeConsumerPropertiesTest( + (binderProps) -> binderProps.setPriorityLevel(0), + (bindingProps) -> bindingProps.setPriorityLevel(0), + expectedProps); } @Test - void basePropSpecifiedWithSameValueAsDefault() { - doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(0), - (bindingProps) -> bindingProps.setPriorityLevel(0), Collections.emptyMap()); + void extPropModifiedAtBinderLevel() { + var expectedProps = defaultExtPropsWith((p) -> p.setReceiverQueueSize(1200)); + doMergeConsumerPropertiesTest( + (binderProps) -> binderProps.setReceiverQueueSize(1200), + SET_NO_PROPS, + expectedProps); } @Test - void extPropSpecifiedAtBinderLevelOnly() { - doMergeConsumerPropertiesTest((binderProps) -> binderProps.setReceiverQueueSize(1200), SET_NO_PROPS, - Map.of("receiverQueueSize", 1200)); + void extPropModifiedAtBindingLevel() { + var expectedProps = defaultExtPropsWith((p) -> p.setReceiverQueueSize(1200)); + doMergeConsumerPropertiesTest( + SET_NO_PROPS, + (bindingProps) -> bindingProps.setReceiverQueueSize(1200), + expectedProps); } @Test - void extPropSpecifiedAtBindingLevelOnly() { - doMergeConsumerPropertiesTest(SET_NO_PROPS, (bindingProps) -> bindingProps.setReceiverQueueSize(1200), - Map.of("receiverQueueSize", 1200)); + void extPropModifiedAtBinderAndBindingLevel() { + var expectedProps = defaultExtPropsWith((p) -> p.setReceiverQueueSize(1200)); + doMergeConsumerPropertiesTest( + (binderProps) -> binderProps.setReceiverQueueSize(1100), + (bindingProps) -> bindingProps.setReceiverQueueSize(1200), + expectedProps); } @Test - void extPropSpecifiedAtBinderAndBindingLevel() { - doMergeConsumerPropertiesTest((binderProps) -> binderProps.setReceiverQueueSize(1100), - (bindingProps) -> bindingProps.setReceiverQueueSize(1200), Map.of("receiverQueueSize", 1200)); - } - - @Test - void extPropSpecifiedWithSameValueAsDefault() { - doMergeConsumerPropertiesTest((binderProps) -> binderProps.setReceiverQueueSize(1000), - (bindingProps) -> bindingProps.setReceiverQueueSize(1000), Collections.emptyMap()); + void extPropModifiedAtBinderAndBindingLevelWithDefaultValue() { + var expectedProps = defaultExtProps(); + doMergeConsumerPropertiesTest( + (binderProps) -> binderProps.setReceiverQueueSize(1000), + (bindingProps) -> bindingProps.setReceiverQueueSize(1000), + expectedProps); } @Test void baseAndExtPropsAreCombined() { - doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(1000), + var expectedProps = defaultExtPropsWith((p) -> p.setReceiverQueueSize(1200)); + expectedProps.put("priorityLevel", 1000); + doMergeConsumerPropertiesTest( + (binderProps) -> binderProps.setPriorityLevel(1000), (bindingProps) -> bindingProps.setReceiverQueueSize(1200), - Map.of("priorityLevel", 1000, "receiverQueueSize", 1200)); + expectedProps); + } + // @formatter:on + + private Map defaultExtProps() { + return new ConsumerConfigProperties().toExtendedConsumerPropertiesMap(); } - void doMergeConsumerPropertiesTest(Consumer binderPropsCustomizer, - Consumer bindingPropsCustomizer, Map expectedProps) { + private Map defaultExtPropsWith(String key, Object value) { + var defaultExtProps = defaultExtProps(); + defaultExtProps.put(key, value); + return defaultExtProps; + } + + private Map defaultExtPropsWith(Consumer extPropsCustomizer) { + var extProps = new ConsumerConfigProperties(); + extPropsCustomizer.accept(extProps); + return extProps.toExtendedConsumerPropertiesMap(); + } + + private void doMergeConsumerPropertiesTest( + Consumer binderPropsCustomizer, + Consumer bindingPropsCustomizer, + Map expectedMergedProperties) { var binderConsumerProps = new ConsumerConfigProperties(); binderPropsCustomizer.accept(binderConsumerProps); var bindingConsumerProps = new ConsumerConfigProperties(); bindingPropsCustomizer.accept(bindingConsumerProps); - var mergedProps = PulsarBinderUtils.mergeModifiedConsumerProperties(binderConsumerProps, - bindingConsumerProps); - assertThat(mergedProps).containsExactlyInAnyOrderEntriesOf(expectedProps); + var mergedProps = PulsarBinderUtils.mergeModifiedConsumerProperties(binderConsumerProps, bindingConsumerProps); + assertThat(mergedProps).containsExactlyInAnyOrderEntriesOf(expectedMergedProperties); } }