SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Pulsar binder for Java 21 (#2918)

Browse Source 
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
This commit is contained in:
Chris Bono
2024-03-14 14:47:21 -05:00
committed by GitHub
parent a2458f3029
commit fcdaf2fdd8
7 changed files with 432 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java
View File

@@ -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.
* <p>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.
* <p>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<String, Object> 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.
* <p>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.
* <p>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<String, Object> 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.
* <p>
* <b>NOTE:</b> 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<String, Object> mergePropertiesWithPrecedence(Map<String, Object> baseProps,
Map<String, Object> binderProps, Map<String, Object> bindingProps) {
Objects.requireNonNull(baseProps, "baseProps must be specified");
static Map<String, Object> mergePropertiesWithPrecedence(Map<String, Object> globalProps,
Map<String, Object> binderProps, Map<String, Object> 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<String, Object> newOrModifiedBinderProps = extractNewOrModifiedProperties(binderProps, baseProps);
Map<String, Object> newOrModifiedBinderProps = extractNewOrModifiedProperties(binderProps, globalProps);
LOGGER.trace(() -> "New or modified binder props: %s".formatted(newOrModifiedBinderProps));
Map<String, Object> newOrModifiedBindingProps = extractNewOrModifiedProperties(bindingProps, baseProps);
Map<String, Object> newOrModifiedBindingProps = extractNewOrModifiedProperties(bindingProps, globalProps);
LOGGER.trace(() -> "New or modified binding props: %s".formatted(newOrModifiedBindingProps));
Map<String, Object> 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
* <a href="https://github.com/apache/pulsar/pull/18344">Pulsar PR</a>.
* @param builder the builder
* @param properties the properties to set on the builder
* @param <T> the payload type
*/
static <T> void loadConf(ProducerBuilder<T> builder, Map<String, Object> properties) {
builder.loadConf(properties);
// Set fields that are not loaded by loadConf
if (properties.containsKey("encryptionKeys")) {
@SuppressWarnings("unchecked")
Collection<String> keys = (Collection<String>) 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"));
}
}
}

3
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java
View File

@@ -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());

93
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ConsumerConfigProperties.java
View File

@@ -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 {
/**

16
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/ProducerConfigProperties.java
View File

@@ -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;
}
/**

91
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/ConsumerConfigPropertiesTests.java
View File

@@ -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<String, String> basePropsInputMap() {
Map<String, String> 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<String, Object> 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);
}

4
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java
View File

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

333
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java
View File

@@ -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<String, Object> baseProps, Map<String, Object> binderProps,
void mergePropertiesTest(String testName, boolean includeDefaults, Map<String, Object> globalProps, Map<String, Object> binderProps,
Map<String, Object> bindingProps, Map<String, Object> expectedMergedProps) {
assertThat(PulsarBinderUtils.mergePropertiesWithPrecedence(baseProps, binderProps, bindingProps))
assertThat(PulsarBinderUtils.mergePropertiesWithPrecedence(globalProps, binderProps, bindingProps, includeDefaults))
.containsExactlyInAnyOrderEntriesOf(expectedMergedProps);
}
// @formatter:off
static Stream<Arguments> 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<String, Object> defaultExtProps() {
return new ProducerConfigProperties().toExtendedProducerPropertiesMap();
}
void doMergeProducerPropertiesTest(Consumer<ProducerConfigProperties> binderPropsCustomizer,
Consumer<ProducerConfigProperties> bindingPropsCustomizer, Map<String, Object> expectedProps) {
private Map<String, Object> defaultExtPropsWith(String key, Object value) {
var defaultExtProps = defaultExtProps();
defaultExtProps.put(key, value);
return defaultExtProps;
}
private Map<String, Object> defaultExtPropsWith(Consumer<ProducerConfigProperties> extPropsCustomizer) {
var extProps = new ProducerConfigProperties();
extPropsCustomizer.accept(extProps);
return extProps.toExtendedProducerPropertiesMap();
}
private void doMergeProducerPropertiesTest(Consumer<ProducerConfigProperties> binderPropsCustomizer,
Consumer<ProducerConfigProperties> bindingPropsCustomizer,
Map<String, Object> 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<String, Object> defaultExtProps() {
return new ConsumerConfigProperties().toExtendedConsumerPropertiesMap();
}
void doMergeConsumerPropertiesTest(Consumer<ConsumerConfigProperties> binderPropsCustomizer,
Consumer<ConsumerConfigProperties> bindingPropsCustomizer, Map<String, Object> expectedProps) {
private Map<String, Object> defaultExtPropsWith(String key, Object value) {
var defaultExtProps = defaultExtProps();
defaultExtProps.put(key, value);
return defaultExtProps;
}
private Map<String, Object> defaultExtPropsWith(Consumer<ConsumerConfigProperties> extPropsCustomizer) {
var extProps = new ConsumerConfigProperties();
extPropsCustomizer.accept(extProps);
return extProps.toExtendedConsumerPropertiesMap();
}
private void doMergeConsumerPropertiesTest(
Consumer<ConsumerConfigProperties> binderPropsCustomizer,
Consumer<ConsumerConfigProperties> bindingPropsCustomizer,
Map<String, Object> 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);
}
}