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); } }