Update Pulsar binder to use new Spring Pulsar starter.

The PulsarBinder relies on the Spring Pulsar Spring Boot starter.
The starter moved out of the spring-pulsar core repo and into
Spring Boot proper. This commit updates the Pulsar binder to
use the new coordinates for the Spring Boot based starter.

* Additionally, the PulsarProperties were greatly reduced in the
  move to Spring Boot. As such, the binder exposes an extended
  set of config properties for producer/consumer (the initial
  set supported before the property reduction).
This commit is contained in:
Chris Bono
2023-09-10 20:31:51 -05:00
committed by Soby Chacko
parent 3c65104b1d
commit db54f5e730
19 changed files with 1535 additions and 521 deletions

View File

@@ -5,7 +5,7 @@
<artifactId>spring-cloud-stream-binder-pulsar</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-pulsar</name>
<description>PUlsar binder implementation</description>
<description>Pulsar binder implementation</description>
<parent>
<groupId>org.springframework.cloud</groupId>
@@ -15,9 +15,9 @@
<dependencies>
<dependency>
<groupId>org.springframework.pulsar</groupId>
<artifactId>spring-pulsar-spring-boot-starter</artifactId>
<version>0.2.1-SNAPSHOT</version>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-pulsar</artifactId>
<version>3.2.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>

View File

@@ -26,7 +26,6 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.pulsar.support.header.PulsarHeaderMapper;
/**
* A delegating {@code PulsarHeaderMapper} that ensures the delegate mapper never includes
* internal binder specific headers during outbound mapping.

View File

@@ -16,20 +16,17 @@
package org.springframework.cloud.stream.binder.pulsar;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import org.springframework.boot.context.properties.PropertyMapper;
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;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.core.log.LogAccessor;
import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
import org.springframework.util.StringUtils;
import org.springframework.util.unit.DataSize;
/**
* Binder utility methods.
@@ -55,12 +52,74 @@ final class PulsarBinderUtils {
* consumer properties
*/
static String subscriptionName(PulsarConsumerProperties consumerProps, ConsumerDestination consumerDestination) {
if (StringUtils.hasText(consumerProps.getSubscriptionName())) {
return consumerProps.getSubscriptionName();
if (StringUtils.hasText(consumerProps.getSubscription().getName())) {
return consumerProps.getSubscription().getName();
}
return SUBSCRIPTION_NAME_FORMAT_STR.formatted(consumerDestination.getName(), UUID.randomUUID());
}
/**
* 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.
* @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
*/
static Map<String, Object> mergeModifiedProducerProperties(ProducerConfigProperties binderProducerProps,
ProducerConfigProperties bindingProducerProps) {
// Layer the base props for common -> binder -> bindings
var baseProducerProps = 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 binderExtProducerProps = binderProducerProps.toExtendedProducerPropertiesMap();
var bindingExtProducerProps = bindingProducerProps.toExtendedProducerPropertiesMap();
var layeredExtProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(extProducerProps,
binderExtProducerProps, bindingExtProducerProps);
// Combine both base and extended layers
var layeredProducerProps = new HashMap<>(layeredBaseProducerProps);
layeredProducerProps.putAll(layeredExtProducerProps);
return layeredProducerProps;
}
/**
* 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.
* @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
*/
static Map<String, Object> mergeModifiedConsumerProperties(ConsumerConfigProperties binderConsumerProps,
ConsumerConfigProperties bindingConsumerProps) {
// Layer the base props for common -> binder -> bindings
var baseConsumerProps = 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 binderExtConsumerProps = binderConsumerProps.toExtendedConsumerPropertiesMap();
var bindingExtConsumerProps = bindingConsumerProps.toExtendedConsumerPropertiesMap();
var layeredExtConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(extConsumerProps,
binderExtConsumerProps, bindingExtConsumerProps);
// Combine both base and extended layers
var layeredConsumerProps = new HashMap<>(layeredBaseConsumerProps);
layeredConsumerProps.putAll(layeredExtConsumerProps);
return layeredConsumerProps;
}
/**
* Merges properties defined at the binder and binding level (binding properties
* override binder properties).
@@ -105,103 +164,4 @@ final class PulsarBinderUtils {
return newOrModifiedProps;
}
/**
* Gets a map representation of a {@link ProducerConfigProperties}.
* @param producerProps the producer props
* @return map representation of producer props where each entry is a field and its
* associated value
*/
static Map<String, Object> convertProducerPropertiesToMap(ProducerConfigProperties producerProps) {
var properties = new PulsarBinderUtils.Properties();
var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(producerProps::getTopicName).to(properties.in("topicName"));
map.from(producerProps::getProducerName).to(properties.in("producerName"));
map.from(producerProps::getSendTimeout).asInt(Duration::toMillis).to(properties.in("sendTimeoutMs"));
map.from(producerProps::getBlockIfQueueFull).to(properties.in("blockIfQueueFull"));
map.from(producerProps::getMaxPendingMessages).to(properties.in("maxPendingMessages"));
map.from(producerProps::getMaxPendingMessagesAcrossPartitions)
.to(properties.in("maxPendingMessagesAcrossPartitions"));
map.from(producerProps::getMessageRoutingMode).to(properties.in("messageRoutingMode"));
map.from(producerProps::getHashingScheme).to(properties.in("hashingScheme"));
map.from(producerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction"));
map.from(producerProps::getBatchingMaxPublishDelay).as(it -> it.toNanos() / 1000)
.to(properties.in("batchingMaxPublishDelayMicros"));
map.from(producerProps::getBatchingPartitionSwitchFrequencyByPublishDelay)
.to(properties.in("batchingPartitionSwitchFrequencyByPublishDelay"));
map.from(producerProps::getBatchingMaxMessages).to(properties.in("batchingMaxMessages"));
map.from(producerProps::getBatchingMaxBytes).asInt(DataSize::toBytes).to(properties.in("batchingMaxBytes"));
map.from(producerProps::getBatchingEnabled).to(properties.in("batchingEnabled"));
map.from(producerProps::getChunkingEnabled).to(properties.in("chunkingEnabled"));
map.from(producerProps::getEncryptionKeys).to(properties.in("encryptionKeys"));
map.from(producerProps::getCompressionType).to(properties.in("compressionType"));
map.from(producerProps::getInitialSequenceId).to(properties.in("initialSequenceId"));
map.from(producerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions"));
map.from(producerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
.to(properties.in("autoUpdatePartitionsIntervalSeconds"));
map.from(producerProps::getMultiSchema).to(properties.in("multiSchema"));
map.from(producerProps::getProducerAccessMode).to(properties.in("accessMode"));
map.from(producerProps::getLazyStartPartitionedProducers).to(properties.in("lazyStartPartitionedProducers"));
map.from(producerProps::getProperties).to(properties.in("properties"));
return properties;
}
/**
* Gets a map representation of a {@link ConsumerConfigProperties}.
* @param consumerProps the consumer props
* @return map representation of consumer props where each entry is a field and its
* associated value
*/
static Map<String, Object> convertConsumerPropertiesToMap(ConsumerConfigProperties consumerProps) {
var properties = new PulsarBinderUtils.Properties();
var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(consumerProps::getTopics).to(properties.in("topicNames"));
map.from(consumerProps::getTopicsPattern).to(properties.in("topicsPattern"));
map.from(consumerProps::getSubscriptionName).to(properties.in("subscriptionName"));
map.from(consumerProps::getSubscriptionType).to(properties.in("subscriptionType"));
map.from(consumerProps::getSubscriptionProperties).to(properties.in("subscriptionProperties"));
map.from(consumerProps::getSubscriptionMode).to(properties.in("subscriptionMode"));
map.from(consumerProps::getReceiverQueueSize).to(properties.in("receiverQueueSize"));
map.from(consumerProps::getAcknowledgementsGroupTime).as(it -> it.toNanos() / 1000)
.to(properties.in("acknowledgementsGroupTimeMicros"));
map.from(consumerProps::getNegativeAckRedeliveryDelay).as(it -> it.toNanos() / 1000)
.to(properties.in("negativeAckRedeliveryDelayMicros"));
map.from(consumerProps::getMaxTotalReceiverQueueSizeAcrossPartitions)
.to(properties.in("maxTotalReceiverQueueSizeAcrossPartitions"));
map.from(consumerProps::getConsumerName).to(properties.in("consumerName"));
map.from(consumerProps::getAckTimeout).as(Duration::toMillis).to(properties.in("ackTimeoutMillis"));
map.from(consumerProps::getTickDuration).as(Duration::toMillis).to(properties.in("tickDurationMillis"));
map.from(consumerProps::getPriorityLevel).to(properties.in("priorityLevel"));
map.from(consumerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction"));
map.from(consumerProps::getProperties).to(properties.in("properties"));
map.from(consumerProps::getReadCompacted).to(properties.in("readCompacted"));
map.from(consumerProps::getSubscriptionInitialPosition).to(properties.in("subscriptionInitialPosition"));
map.from(consumerProps::getPatternAutoDiscoveryPeriod).to(properties.in("patternAutoDiscoveryPeriod"));
map.from(consumerProps::getRegexSubscriptionMode).to(properties.in("regexSubscriptionMode"));
map.from(consumerProps::getDeadLetterPolicy).to(properties.in("deadLetterPolicy"));
map.from(consumerProps::getRetryEnable).to(properties.in("retryEnable"));
map.from(consumerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions"));
map.from(consumerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
.to(properties.in("autoUpdatePartitionsIntervalSeconds"));
map.from(consumerProps::getReplicateSubscriptionState).to(properties.in("replicateSubscriptionState"));
map.from(consumerProps::getResetIncludeHead).to(properties.in("resetIncludeHead"));
map.from(consumerProps::getBatchIndexAckEnabled).to(properties.in("batchIndexAckEnabled"));
map.from(consumerProps::getAckReceiptEnabled).to(properties.in("ackReceiptEnabled"));
map.from(consumerProps::getPoolMessages).to(properties.in("poolMessages"));
map.from(consumerProps::getStartPaused).to(properties.in("startPaused"));
map.from(consumerProps::getAutoAckOldestChunkedMessageOnQueueFull)
.to(properties.in("autoAckOldestChunkedMessageOnQueueFull"));
map.from(consumerProps::getMaxPendingChunkedMessage).to(properties.in("maxPendingChunkedMessage"));
map.from(consumerProps::getExpireTimeOfIncompleteChunkedMessage).as(Duration::toMillis)
.to(properties.in("expireTimeOfIncompleteChunkedMessageMillis"));
return properties;
}
static class Properties extends HashMap<String, Object> {
<V> java.util.function.Consumer<V> in(String key) {
return (value) -> put(key, value);
}
}
}

View File

@@ -47,8 +47,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.autoconfigure.ConsumerConfigProperties;
import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
import org.springframework.pulsar.core.ProducerBuilderConfigurationUtil;
import org.springframework.pulsar.core.ProducerBuilderCustomizer;
import org.springframework.pulsar.core.PulsarConsumerFactory;
@@ -61,7 +59,6 @@ import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.listener.PulsarRecordMessageListener;
import org.springframework.pulsar.support.header.PulsarHeaderMapper;
/**
* {@link Binder} implementation for Apache Pulsar.
*
@@ -112,15 +109,10 @@ public class PulsarMessageChannelBinder extends
schema = null;
}
var baseProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(new ProducerConfigProperties());
var binderProducerProps = PulsarBinderUtils
.convertProducerPropertiesToMap(this.binderConfigProps.getProducer());
var bindingProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(producerProperties.getExtension());
var mergedProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseProducerProps,
binderProducerProps, bindingProducerProps);
var layeredProducerProps = PulsarBinderUtils.mergeModifiedProducerProperties(
this.binderConfigProps.getProducer(), producerProperties.getExtension());
var handler = new PulsarProducerConfigurationMessageHandler(this.pulsarTemplate, schema, destination.getName(),
(builder) -> ProducerBuilderConfigurationUtil.loadConf(builder, mergedProducerProps),
(builder) -> ProducerBuilderConfigurationUtil.loadConf(builder, layeredProducerProps),
determineOutboundHeaderMapper(producerProperties));
handler.setApplicationContext(getApplicationContext());
handler.setBeanFactory(getBeanFactory());
@@ -167,13 +159,9 @@ public class PulsarMessageChannelBinder extends
var subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination);
containerProperties.setSubscriptionName(subscriptionName);
var baseConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(new ConsumerConfigProperties());
var binderConsumerProps = PulsarBinderUtils
.convertConsumerPropertiesToMap(this.binderConfigProps.getConsumer());
var bindingConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getExtension());
var mergedConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseConsumerProps,
binderConsumerProps, bindingConsumerProps);
containerProperties.getPulsarConsumerProperties().putAll(mergedConsumerProps);
var layeredConsumerProps = PulsarBinderUtils
.mergeModifiedConsumerProperties(this.binderConfigProps.getConsumer(), properties.getExtension());
containerProperties.getPulsarConsumerProperties().putAll(layeredConsumerProps);
containerProperties.updateContainerProperties();
var container = new DefaultPulsarMessageListenerContainer<>(this.pulsarConsumerFactory, containerProperties);

View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.stream.binder.pulsar.config;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.pulsar.PulsarMessageChannelBinder;
@@ -25,7 +26,6 @@ import org.springframework.cloud.stream.binder.pulsar.properties.PulsarExtendedB
import org.springframework.cloud.stream.binder.pulsar.provisioning.PulsarTopicProvisioner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.autoconfigure.PulsarProperties;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
@@ -48,7 +48,7 @@ public class PulsarBinderConfiguration {
@Bean
public PulsarTopicProvisioner pulsarTopicProvisioner(PulsarAdministration pulsarAdministration,
PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties) {
PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties) {
return new PulsarTopicProvisioner(pulsarAdministration, pulsarBinderConfigurationProperties);
}
@@ -63,9 +63,9 @@ public class PulsarBinderConfiguration {
@Bean
public PulsarMessageChannelBinder pulsarMessageChannelBinder(PulsarTopicProvisioner pulsarTopicProvisioner,
PulsarTemplate<Object> pulsarTemplate, PulsarConsumerFactory<byte[]> pulsarConsumerFactory,
PulsarBinderConfigurationProperties binderConfigProps, PulsarExtendedBindingProperties bindingConfigProps,
SchemaResolver schemaResolver, PulsarHeaderMapper headerMapper) {
PulsarTemplate<Object> pulsarTemplate, PulsarConsumerFactory<?> pulsarConsumerFactory,
PulsarBinderConfigurationProperties binderConfigProps, PulsarExtendedBindingProperties bindingConfigProps,
SchemaResolver schemaResolver, PulsarHeaderMapper headerMapper) {
PulsarMessageChannelBinder pulsarMessageChannelBinder = new PulsarMessageChannelBinder(pulsarTopicProvisioner,
pulsarTemplate, pulsarConsumerFactory, binderConfigProps, schemaResolver, headerMapper);
pulsarMessageChannelBinder.setExtendedBindingProperties(bindingConfigProps);

View File

@@ -0,0 +1,447 @@
/*
* Copyright 2023-2023 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.pulsar.properties;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.MessageId;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.util.Assert;
/**
* Configuration properties used to specify Pulsar consumers.
*
* @author Chris Bono
*/
public class ConsumerConfigProperties extends PulsarProperties.Consumer {
private final Acknowledgement ack = new Acknowledgement();
private final Chunking chunk = new Chunking();
private final Subscription subscription = new Subscription();
/**
* Number of messages that can be accumulated before the consumer calls "receive".
*/
private Integer receiverQueueSize = 1000;
/**
* Maximum number of messages that a consumer can be pushed at once from a broker
* across all partitions.
*/
private Integer maxTotalReceiverQueueSizeAcrossPartitions = 50000;
/**
* Action the consumer will take in case of decryption failure.
*/
private ConsumerCryptoFailureAction cryptoFailureAction = ConsumerCryptoFailureAction.FAIL;
/**
* Map of properties to add to the consumer.
*/
private SortedMap<String, String> properties = new TreeMap<>();
/**
* Auto-discovery period for topics when topic pattern is used in minutes.
*/
private Integer patternAutoDiscoveryPeriod = 1;
/**
* Whether the consumer auto-subscribes for partition increase. This is only for
* partitioned consumers.
*/
private Boolean autoUpdatePartitions = true;
/**
* Interval of partitions discovery updates.
*/
private Duration autoUpdatePartitionsInterval = Duration.ofMinutes(1);
/**
* Whether to include the given position of any reset operation like
* {@link org.apache.pulsar.client.api.Consumer#seek(long) or
* {@link ConsumerConfigProperties#seek(MessageId)}}.
*/
private Boolean resetIncludeHead = false;
/**
* Whether pooling of messages and the underlying data buffers is enabled.
*/
private Boolean poolMessages = false;
/**
* Whether to start the consumer in a paused state.
*/
private Boolean startPaused = false;
public Acknowledgement getAck() {
return this.ack;
}
public Chunking getChunk() {
return this.chunk;
}
public Subscription getSubscription() {
return this.subscription;
}
public Integer getReceiverQueueSize() {
return this.receiverQueueSize;
}
public void setReceiverQueueSize(Integer receiverQueueSize) {
this.receiverQueueSize = receiverQueueSize;
}
public Integer getMaxTotalReceiverQueueSizeAcrossPartitions() {
return this.maxTotalReceiverQueueSizeAcrossPartitions;
}
public void setMaxTotalReceiverQueueSizeAcrossPartitions(Integer maxTotalReceiverQueueSizeAcrossPartitions) {
this.maxTotalReceiverQueueSizeAcrossPartitions = maxTotalReceiverQueueSizeAcrossPartitions;
}
public ConsumerCryptoFailureAction getCryptoFailureAction() {
return this.cryptoFailureAction;
}
public void setCryptoFailureAction(ConsumerCryptoFailureAction cryptoFailureAction) {
this.cryptoFailureAction = cryptoFailureAction;
}
public SortedMap<String, String> getProperties() {
return this.properties;
}
public void setProperties(SortedMap<String, String> properties) {
this.properties = properties;
}
public Integer getPatternAutoDiscoveryPeriod() {
return this.patternAutoDiscoveryPeriod;
}
public void setPatternAutoDiscoveryPeriod(Integer patternAutoDiscoveryPeriod) {
this.patternAutoDiscoveryPeriod = patternAutoDiscoveryPeriod;
}
public Boolean getAutoUpdatePartitions() {
return this.autoUpdatePartitions;
}
public void setAutoUpdatePartitions(Boolean autoUpdatePartitions) {
this.autoUpdatePartitions = autoUpdatePartitions;
}
public Duration getAutoUpdatePartitionsInterval() {
return this.autoUpdatePartitionsInterval;
}
public void setAutoUpdatePartitionsInterval(Duration autoUpdatePartitionsInterval) {
this.autoUpdatePartitionsInterval = autoUpdatePartitionsInterval;
}
public Boolean getResetIncludeHead() {
return this.resetIncludeHead;
}
public void setResetIncludeHead(Boolean resetIncludeHead) {
this.resetIncludeHead = resetIncludeHead;
}
public Boolean getPoolMessages() {
return this.poolMessages;
}
public void setPoolMessages(Boolean poolMessages) {
this.poolMessages = poolMessages;
}
public Boolean getStartPaused() {
return this.startPaused;
}
public void setStartPaused(Boolean startPaused) {
this.startPaused = startPaused;
}
/**
* Gets a map representation of the base consumer properties (those defined in parent
* class).
* @return map of base consumer properties and associated values.
*/
public Map<String, Object> toBaseConsumerPropertiesMap() {
var consumerProps = new ConsumerConfigProperties.Properties();
var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(this::getDeadLetterPolicy).as(this::toPulsarDeadLetterPolicy).to(consumerProps.in("deadLetterPolicy"));
map.from(this::getName).to(consumerProps.in("consumerName"));
map.from(this::getPriorityLevel).to(consumerProps.in("priorityLevel"));
map.from(this::isReadCompacted).to(consumerProps.in("readCompacted"));
map.from(this::isRetryEnable).to(consumerProps.in("retryEnable"));
map.from(this::getTopics).to(consumerProps.in("topicNames"));
map.from(this::getTopicsPattern).to(consumerProps.in("topicsPattern"));
return consumerProps;
}
/**
* Gets a map representation of the extended consumer properties (those defined in
* this class).
* @return map of extended consumer properties and associated values.
*/
public Map<String, Object> toExtendedConsumerPropertiesMap() {
var consumerProps = new ConsumerConfigProperties.Properties();
var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(this::getAutoUpdatePartitions).to(consumerProps.in("autoUpdatePartitions"));
map.from(this::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
.to(consumerProps.in("autoUpdatePartitionsIntervalSeconds"));
map.from(this::getCryptoFailureAction).to(consumerProps.in("cryptoFailureAction"));
map.from(this::getMaxTotalReceiverQueueSizeAcrossPartitions)
.to(consumerProps.in("maxTotalReceiverQueueSizeAcrossPartitions"));
map.from(this::getPatternAutoDiscoveryPeriod).to(consumerProps.in("patternAutoDiscoveryPeriod"));
map.from(this::getPoolMessages).to(consumerProps.in("poolMessages"));
map.from(this::getProperties).to(consumerProps.in("properties"));
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"));
return consumerProps;
}
/**
* Gets a map representation of base and extended consumer properties.
* @return map of base and extended consumer properties and associated values.
*/
public Map<String, Object> toAllConsumerPropertiesMap() {
var consumerProps = this.toBaseConsumerPropertiesMap();
consumerProps.putAll(this.toExtendedConsumerPropertiesMap());
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 {
/**
* Whether the batching index acknowledgment is enabled.
*/
private Boolean batchIndexEnabled = false;
/**
* Time to group acknowledgements before sending them to the broker.
*/
private Duration groupTime = Duration.ofMillis(100);
/**
* Whether an acknowledgement receipt is enabled.
*/
private Boolean receiptEnabled = false;
/**
* Delay before re-delivering messages that have failed to be processed.
*/
private Duration redeliveryDelay = Duration.ofMinutes(1);
/**
* Timeout for unacked messages to be redelivered.
*/
private Duration timeout = Duration.ZERO;
/**
* Precision for the ack timeout messages tracker.
*/
private Duration timeoutTickDuration = Duration.ofSeconds(1);
public Boolean getBatchIndexEnabled() {
return this.batchIndexEnabled;
}
public void setBatchIndexEnabled(Boolean batchIndexEnabled) {
this.batchIndexEnabled = batchIndexEnabled;
}
public Duration getGroupTime() {
return this.groupTime;
}
public void setGroupTime(Duration groupTime) {
this.groupTime = groupTime;
}
public Boolean getReceiptEnabled() {
return this.receiptEnabled;
}
public void setReceiptEnabled(Boolean receiptEnabled) {
this.receiptEnabled = receiptEnabled;
}
public Duration getRedeliveryDelay() {
return this.redeliveryDelay;
}
public void setRedeliveryDelay(Duration redeliveryDelay) {
this.redeliveryDelay = redeliveryDelay;
}
public Duration getTimeout() {
return this.timeout;
}
public void setTimeout(Duration timeout) {
this.timeout = timeout;
}
public Duration getTimeoutTickDuration() {
return this.timeoutTickDuration;
}
public void setTimeoutTickDuration(Duration timeoutTickDuration) {
this.timeoutTickDuration = timeoutTickDuration;
}
}
public static class Chunking {
/**
* Whether to automatically drop outstanding uncompleted chunked messages once the
* consumer queue reaches the threshold set by the 'maxPendingMessages' property.
*/
private Boolean autoAckOldestOnQueueFull = true;
/**
* The maximum time period for a consumer to receive all chunks of a message - if
* this threshold is exceeded the consumer will expire the incomplete chunks.
*/
private Duration expireTimeIncomplete = Duration.ofMinutes(1);
/**
* Maximum number of chunked messages to be kept in memory.
*/
private Integer maxPendingMessages = 10;
public Boolean getAutoAckOldestOnQueueFull() {
return this.autoAckOldestOnQueueFull;
}
public void setAutoAckOldestOnQueueFull(Boolean autoAckOldestOnQueueFull) {
this.autoAckOldestOnQueueFull = autoAckOldestOnQueueFull;
}
public Duration getExpireTimeIncomplete() {
return this.expireTimeIncomplete;
}
public void setExpireTimeIncomplete(Duration expireTimeIncomplete) {
this.expireTimeIncomplete = expireTimeIncomplete;
}
public Integer getMaxPendingMessages() {
return this.maxPendingMessages;
}
public void setMaxPendingMessages(Integer maxPendingMessages) {
this.maxPendingMessages = maxPendingMessages;
}
}
public static class Subscription extends PulsarProperties.Consumer.Subscription {
/**
* Map of properties to add to the subscription.
*/
private Map<String, String> properties = new HashMap<>();
/**
* Whether to replicate subscription state.
*/
private Boolean replicateState = false;
public Map<String, String> getProperties() {
return this.properties;
}
public void setProperties(Map<String, String> properties) {
this.properties = properties;
}
public Boolean getReplicateState() {
return this.replicateState;
}
public void setReplicateState(Boolean replicateState) {
this.replicateState = replicateState;
}
}
static class Properties extends HashMap<String, Object> {
<V> java.util.function.Consumer<V> in(String key) {
return (value) -> put(key, value);
}
}
}

View File

@@ -0,0 +1,333 @@
/*
* Copyright 2023-2023 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.pulsar.properties;
import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.lang.Nullable;
import org.springframework.util.unit.DataSize;
/**
* Configuration properties used to specify Pulsar producers.
*
* @author Chris Bono
*/
public class ProducerConfigProperties extends PulsarProperties.Producer {
/**
* Whether the "send" and "sendAsync" methods should block if the outgoing message
* queue is full.
*/
private Boolean blockIfQueueFull = false;
/**
* Maximum number of pending messages for the producer.
*/
private Integer maxPendingMessages = 1000;
/**
* Maximum number of pending messages across all the partitions.
*/
private Integer maxPendingMessagesAcrossPartitions = 50000;
/**
* Action the producer will take in case of encryption failure.
*/
private ProducerCryptoFailureAction cryptoFailureAction = ProducerCryptoFailureAction.FAIL;
/**
* Names of the public encryption keys to use when encrypting data.
*/
private Set<String> encryptionKeys = new HashSet<>();
/**
* Baseline for the sequence ids for messages published by the producer.
*/
@Nullable
private Long initialSequenceId;
/**
* Whether partitioned producer automatically discover new partitions at runtime.
*/
private Boolean autoUpdatePartitions = true;
/**
* Interval of partitions discovery updates.
*/
private Duration autoUpdatePartitionsInterval = Duration.ofMinutes(1);
/**
* Whether the multiple schema mode is enabled.
*/
private Boolean multiSchema = true;
/**
* Whether producers in Shared mode register and connect immediately to the owner
* broker of each partition or start lazily on demand.
*/
private Boolean lazyStartPartitionedProducers = false;
private final Batching batch = new Batching();
public Batching getBatch() {
return this.batch;
}
/**
* Map of properties to add to the producer.
*/
private Map<String, String> properties = new HashMap<>();
public Boolean getBlockIfQueueFull() {
return this.blockIfQueueFull;
}
public void setBlockIfQueueFull(Boolean blockIfQueueFull) {
this.blockIfQueueFull = blockIfQueueFull;
}
public Integer getMaxPendingMessages() {
return this.maxPendingMessages;
}
public void setMaxPendingMessages(Integer maxPendingMessages) {
this.maxPendingMessages = maxPendingMessages;
}
public Integer getMaxPendingMessagesAcrossPartitions() {
return this.maxPendingMessagesAcrossPartitions;
}
public void setMaxPendingMessagesAcrossPartitions(Integer maxPendingMessagesAcrossPartitions) {
this.maxPendingMessagesAcrossPartitions = maxPendingMessagesAcrossPartitions;
}
public ProducerCryptoFailureAction getCryptoFailureAction() {
return this.cryptoFailureAction;
}
public void setCryptoFailureAction(ProducerCryptoFailureAction cryptoFailureAction) {
this.cryptoFailureAction = cryptoFailureAction;
}
public Set<String> getEncryptionKeys() {
return this.encryptionKeys;
}
public void setEncryptionKeys(Set<String> encryptionKeys) {
this.encryptionKeys = encryptionKeys;
}
@Nullable
public Long getInitialSequenceId() {
return this.initialSequenceId;
}
public void setInitialSequenceId(@Nullable Long initialSequenceId) {
this.initialSequenceId = initialSequenceId;
}
public Boolean getAutoUpdatePartitions() {
return this.autoUpdatePartitions;
}
public void setAutoUpdatePartitions(Boolean autoUpdatePartitions) {
this.autoUpdatePartitions = autoUpdatePartitions;
}
public Duration getAutoUpdatePartitionsInterval() {
return this.autoUpdatePartitionsInterval;
}
public void setAutoUpdatePartitionsInterval(Duration autoUpdatePartitionsInterval) {
this.autoUpdatePartitionsInterval = autoUpdatePartitionsInterval;
}
public Boolean getMultiSchema() {
return this.multiSchema;
}
public void setMultiSchema(Boolean multiSchema) {
this.multiSchema = multiSchema;
}
public Boolean getLazyStartPartitionedProducers() {
return this.lazyStartPartitionedProducers;
}
public void setLazyStartPartitionedProducers(Boolean lazyStartPartitionedProducers) {
this.lazyStartPartitionedProducers = lazyStartPartitionedProducers;
}
public Map<String, String> getProperties() {
return this.properties;
}
public void setProperties(Map<String, String> properties) {
this.properties = properties;
}
/**
* Gets a map representation of the base producer properties (those defined in parent
* class).
* @return map of base producer properties and associated values.
*/
public Map<String, Object> toBaseProducerPropertiesMap() {
var producerProps = new ProducerConfigProperties.Properties();
var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(this::getAccessMode).to(producerProps.in("accessMode"));
map.from(this::isBatchingEnabled).to(producerProps.in("batchingEnabled"));
map.from(this::isChunkingEnabled).to(producerProps.in("chunkingEnabled"));
map.from(this::getCompressionType).to(producerProps.in("compressionType"));
map.from(this::getHashingScheme).to(producerProps.in("hashingScheme"));
map.from(this::getMessageRoutingMode).to(producerProps.in("messageRoutingMode"));
map.from(this::getName).to(producerProps.in("producerName"));
map.from(this::getSendTimeout).asInt(Duration::toMillis).to(producerProps.in("sendTimeoutMs"));
map.from(this::getTopicName).to(producerProps.in("topicName"));
return producerProps;
}
/**
* Gets a map representation of the extended producer properties (those defined in
* this class).
* @return map of extended producer properties and associated values.
*/
public Map<String, Object> toExtendedProducerPropertiesMap() {
var producerProps = new ProducerConfigProperties.Properties();
var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(this::getAutoUpdatePartitions).to(producerProps.in("autoUpdatePartitions"));
map.from(this::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
.to(producerProps.in("autoUpdatePartitionsIntervalSeconds"));
map.from(this::getBlockIfQueueFull).to(producerProps.in("blockIfQueueFull"));
map.from(this::getCryptoFailureAction).to(producerProps.in("cryptoFailureAction"));
map.from(this::getEncryptionKeys).to(producerProps.in("encryptionKeys"));
map.from(this::getInitialSequenceId).to(producerProps.in("initialSequenceId"));
map.from(this::getLazyStartPartitionedProducers).to(producerProps.in("lazyStartPartitionedProducers"));
map.from(this::getMaxPendingMessages).to(producerProps.in("maxPendingMessages"));
map.from(this::getMaxPendingMessagesAcrossPartitions)
.to(producerProps.in("maxPendingMessagesAcrossPartitions"));
map.from(this::getMultiSchema).to(producerProps.in("multiSchema"));
map.from(this::getProperties).to(producerProps.in("properties"));
if (this.isBatchingEnabled()) {
map.from(this::getBatch).as(Batching::getMaxPublishDelay).as(it -> it.toNanos() / 1000)
.to(producerProps.in("batchingMaxPublishDelayMicros"));
map.from(this::getBatch).as(Batching::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)
.to(producerProps.in("batchingMaxBytes"));
}
return producerProps;
}
/**
* Gets a map representation of base and extended producer properties.
* @return map of base and extended producer properties and associated values.
*/
public Map<String, Object> toAllProducerPropertiesMap() {
var producerProps = this.toBaseProducerPropertiesMap();
producerProps.putAll(this.toExtendedProducerPropertiesMap());
return producerProps;
}
public static class Batching {
/**
* Time period within which the messages sent will be batched.
*/
private Duration maxPublishDelay = Duration.ofMillis(1);
/**
* Partition switch frequency while batching of messages is enabled and using
* round-robin routing mode for non-keyed message.
*/
private Integer partitionSwitchFrequencyByPublishDelay = 10;
/**
* Maximum number of messages to be batched.
*/
private Integer maxMessages = 1000;
/**
* Maximum number of bytes permitted in a batch.
*/
private DataSize maxBytes = DataSize.ofKilobytes(128);
/**
* Whether to automatically batch messages.
*/
private Boolean enabled = true;
public Duration getMaxPublishDelay() {
return this.maxPublishDelay;
}
public void setMaxPublishDelay(Duration maxPublishDelay) {
this.maxPublishDelay = maxPublishDelay;
}
public Integer getPartitionSwitchFrequencyByPublishDelay() {
return this.partitionSwitchFrequencyByPublishDelay;
}
public void setPartitionSwitchFrequencyByPublishDelay(Integer partitionSwitchFrequencyByPublishDelay) {
this.partitionSwitchFrequencyByPublishDelay = partitionSwitchFrequencyByPublishDelay;
}
public Integer getMaxMessages() {
return this.maxMessages;
}
public void setMaxMessages(Integer maxMessages) {
this.maxMessages = maxMessages;
}
public DataSize getMaxBytes() {
return this.maxBytes;
}
public void setMaxBytes(DataSize maxBytes) {
this.maxBytes = maxBytes;
}
public Boolean getEnabled() {
return this.enabled;
}
public void setEnabled(Boolean enabled) {
this.enabled = enabled;
}
}
static class Properties extends HashMap<String, Object> {
<V> java.util.function.Consumer<V> in(String key) {
return (value) -> put(key, value);
}
}
}

View File

@@ -19,8 +19,6 @@ package org.springframework.cloud.stream.binder.pulsar.properties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
/**
* {@link ConfigurationProperties @ConfigurationProperties} for the Pulsar binder.

View File

@@ -19,7 +19,6 @@ package org.springframework.cloud.stream.binder.pulsar.properties;
import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
/**
* Pulsar consumer properties used by the binder.

View File

@@ -22,7 +22,6 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
/**
* {@link ConfigurationProperties @ConfigurationProperties} for Pulsar binder specific
* extensions to the common binding properties.

View File

@@ -19,7 +19,6 @@ package org.springframework.cloud.stream.binder.pulsar.properties;
import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
/**
* Pulsar producer properties used by the binder.

View File

@@ -0,0 +1,180 @@
/*
* Copyright 2023-2023 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.pulsar;
import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Test;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
import org.springframework.cloud.stream.binder.pulsar.properties.ConsumerConfigProperties;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
/**
* Unit tests for {@link ConsumerConfigProperties}.
*
* @author Soby Chacko
* @author Chris Bono
*/
public class ConsumerConfigPropertiesTests {
@Test
void basePropsCanBeExtractedToMap() {
var inputProps = basePropsInputMap();
var consumerConfigProps = bindInputPropsToConsumerConfigProps(inputProps);
var outputProps = consumerConfigProps.toBaseConsumerPropertiesMap();
verifyOutputPropsCanBeLoadedInConsumerBuilder(outputProps);
verifyBasePropsInOutputMap(outputProps);
}
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");
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))
.containsExactly("my-topic"))
.hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern"));
}
@Test
void extendedPropsCanBeExtractedToMap() {
var inputProps = basePropsInputMap();
inputProps.putAll(extendedPropsInputMap());
var consumerConfigProps = bindInputPropsToConsumerConfigProps(inputProps);
var outputProps = consumerConfigProps.toExtendedConsumerPropertiesMap();
verifyOutputPropsCanBeLoadedInConsumerBuilder(outputProps);
verifyExtendedPropsInOutputMap(outputProps);
}
@Test
void allPropsCanBeExtractedToMap() {
var inputProps = basePropsInputMap();
inputProps.putAll(extendedPropsInputMap());
var consumerConfigProps = bindInputPropsToConsumerConfigProps(inputProps);
var outputProps = consumerConfigProps.toAllConsumerPropertiesMap();
verifyOutputPropsCanBeLoadedInConsumerBuilder(outputProps);
verifyBasePropsInOutputMap(outputProps);
verifyExtendedPropsInOutputMap(outputProps);
}
private Map<String, String> extendedPropsInputMap() {
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.pulsar.consumer.auto-update-partitions", "true");
inputProps.put("spring.pulsar.consumer.auto-update-partitions-interval", "10s");
inputProps.put("spring.pulsar.consumer.crypto-failure-action", "discard");
inputProps.put("spring.pulsar.consumer.max-total-receiver-queue-size-across-partitions", "5");
inputProps.put("spring.pulsar.consumer.pattern-auto-discovery-period", "9");
inputProps.put("spring.pulsar.consumer.pool-messages", "true");
inputProps.put("spring.pulsar.consumer.properties[my-prop]", "my-prop-value");
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");
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");
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");
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;
}
private void verifyExtendedPropsInOutputMap(Map<String, Object> outputProps) {
assertThat(outputProps).containsEntry("autoUpdatePartitions", true)
.containsEntry("autoUpdatePartitionsIntervalSeconds", 10L)
.containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD)
.containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5)
.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("ackReceiptEnabled", true)
.containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L)
.containsEntry("autoAckOldestChunkedMessageOnQueueFull", false)
.containsEntry("maxPendingChunkedMessage", 11).containsEntry("subscriptionName", "my-subscription")
.containsEntry("subscriptionType", SubscriptionType.Shared)
.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);
}
private void verifyOutputPropsCanBeLoadedInConsumerBuilder(Map<String, Object> outputProps) {
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(outputProps,
new ConsumerConfigurationData<>(), ConsumerConfigurationData.class));
}
private ConsumerConfigProperties bindInputPropsToConsumerConfigProps(Map<String, String> inputProps) {
return new Binder(new MapConfigurationPropertySource(inputProps))
.bind("spring.pulsar.consumer", Bindable.ofInstance(new ConsumerConfigProperties())).get();
}
}

View File

@@ -0,0 +1,163 @@
/*
* Copyright 2023-2023 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.pulsar;
import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.HashingScheme;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.ProducerAccessMode;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Test;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
import org.springframework.cloud.stream.binder.pulsar.properties.ProducerConfigProperties;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
/**
* Unit tests for {@link ProducerConfigProperties}.
*
* @author Soby Chacko
* @author Chris Bono
*/
class ProducerConfigPropertiesTests {
@Test
void basePropsCanBeExtractedToMap() {
var inputProps = basePropsInputMap();
var producerConfigProps = bindInputPropsToProducerConfigProps(inputProps);
var outputProps = producerConfigProps.toBaseProducerPropertiesMap();
verifyOutputPropsCanBeLoadedInProducerBuilder(outputProps);
verifyBasePropsInOutputMap(outputProps);
}
private Map<String, String> basePropsInputMap() {
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.pulsar.producer.access-mode", "Exclusive");
inputProps.put("spring.pulsar.producer.batching-enabled", "true");
inputProps.put("spring.pulsar.producer.chunking-enabled", "true");
inputProps.put("spring.pulsar.producer.compression-type", "lz4");
inputProps.put("spring.pulsar.producer.hashing-scheme", "murmur3_32hash");
inputProps.put("spring.pulsar.producer.message-routing-mode", "custompartition");
inputProps.put("spring.pulsar.producer.name", "my-producer");
inputProps.put("spring.pulsar.producer.send-timeout", "2s");
inputProps.put("spring.pulsar.producer.topic-name", "my-topic");
return inputProps;
}
private void verifyBasePropsInOutputMap(Map<String, Object> outputProps) {
assertThat(outputProps).containsEntry("accessMode", ProducerAccessMode.Exclusive)
.containsEntry("batchingEnabled", true).containsEntry("chunkingEnabled", true)
.containsEntry("compressionType", CompressionType.LZ4)
.containsEntry("hashingScheme", HashingScheme.Murmur3_32Hash)
.containsEntry("messageRoutingMode", MessageRoutingMode.CustomPartition)
.containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000)
.containsEntry("topicName", "my-topic");
}
@Test
void extendedPropsCanBeExtractedToMap() {
var inputProps = basePropsInputMap();
inputProps.putAll(extendedPropsInputMap());
var producerConfigProps = bindInputPropsToProducerConfigProps(inputProps);
var outputProps = producerConfigProps.toExtendedProducerPropertiesMap();
verifyOutputPropsCanBeLoadedInProducerBuilder(outputProps);
verifyExtendedPropsInOutputMap(outputProps);
}
@Test
void allPropsCanBeExtractedToMap() {
var inputProps = basePropsInputMap();
inputProps.putAll(extendedPropsInputMap());
var producerConfigProps = bindInputPropsToProducerConfigProps(inputProps);
var outputProps = producerConfigProps.toAllProducerPropertiesMap();
verifyOutputPropsCanBeLoadedInProducerBuilder(outputProps);
verifyBasePropsInOutputMap(outputProps);
verifyExtendedPropsInOutputMap(outputProps);
}
@Test
void batchPropsSkippedWhenBatchDisabled() {
var inputProps = basePropsInputMap();
inputProps.putAll(extendedPropsInputMap());
inputProps.put("spring.pulsar.producer.batching-enabled", "false");
var producerConfigProps = bindInputPropsToProducerConfigProps(inputProps);
var outputProps = producerConfigProps.toAllProducerPropertiesMap();
assertThat(outputProps).doesNotContainKey("batchingMaxPublishDelayMicros")
.doesNotContainKey("batchingPartitionSwitchFrequencyByPublishDelay")
.doesNotContainKey("batchingMaxMessages").doesNotContainKey("batchingMaxBytes");
}
private Map<String, String> extendedPropsInputMap() {
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.pulsar.producer.auto-update-partitions", "true");
inputProps.put("spring.pulsar.producer.auto-update-partitions-interval", "15s");
inputProps.put("spring.pulsar.producer.block-if-queue-full", "true");
inputProps.put("spring.pulsar.producer.crypto-failure-action", "send");
inputProps.put("spring.pulsar.producer.encryption-keys[0]", "my-key");
inputProps.put("spring.pulsar.producer.initial-sequence-id", "9");
inputProps.put("spring.pulsar.producer.lazy-start=partitioned-producers", "true");
inputProps.put("spring.pulsar.producer.max-pending-messages", "3");
inputProps.put("spring.pulsar.producer.max-pending-messages-across-partitions", "4");
inputProps.put("spring.pulsar.producer.multi-schema", "true");
inputProps.put("spring.pulsar.producer.properties[my-prop]", "my-prop-value");
inputProps.put("spring.pulsar.producer.batch.max-publish-delay", "5s");
inputProps.put("spring.pulsar.producer.batch.partition-switch-frequency-by-publish-delay", "6");
inputProps.put("spring.pulsar.producer.batch.max-messages", "7");
inputProps.put("spring.pulsar.producer.batch.max-bytes", "8");
return inputProps;
}
private void verifyExtendedPropsInOutputMap(Map<String, Object> outputProps) {
assertThat(outputProps).containsEntry("autoUpdatePartitions", true)
.containsEntry("autoUpdatePartitionsIntervalSeconds", 15L).containsEntry("blockIfQueueFull", true)
.containsEntry("cryptoFailureAction", ProducerCryptoFailureAction.SEND)
.hasEntrySatisfying("encryptionKeys",
keys -> assertThat(keys).asInstanceOf(InstanceOfAssertFactories.collection(String.class))
.containsExactly("my-key"))
.containsEntry("initialSequenceId", 9L).containsEntry("lazyStartPartitionedProducers", true)
.containsEntry("maxPendingMessages", 3).containsEntry("maxPendingMessagesAcrossPartitions", 4)
.containsEntry("multiSchema", true)
.hasEntrySatisfying("properties",
properties -> assertThat(properties)
.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
.containsEntry("my-prop", "my-prop-value"))
.containsEntry("batchingMaxPublishDelayMicros", 5_000_000L)
.containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6)
.containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8);
}
private void verifyOutputPropsCanBeLoadedInProducerBuilder(Map<String, Object> outputProps) {
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(outputProps,
new ProducerConfigurationData(), ProducerConfigurationData.class));
}
private ProducerConfigProperties bindInputPropsToProducerConfigProps(Map<String, String> inputProps) {
return new Binder(new MapConfigurationPropertySource(inputProps))
.bind("spring.pulsar.producer", Bindable.ofInstance(new ProducerConfigProperties())).get();
}
}

View File

@@ -29,7 +29,6 @@ import org.junit.jupiter.api.Test;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.ConfigurationPropertySource;
import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
import org.springframework.cloud.stream.binder.pulsar.properties.PulsarBinderConfigurationProperties;
@@ -43,79 +42,57 @@ import static org.assertj.core.api.Assertions.assertThatNoException;
*/
public class PulsarBinderConfigurationPropertiesTests {
private final PulsarBinderConfigurationProperties properties = new PulsarBinderConfigurationProperties();
private void bind(Map<String, String> map) {
ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
new Binder(source).bind("spring.cloud.stream.pulsar.binder", Bindable.ofInstance(this.properties));
}
@Test
void partitionCountProperty() {
assertThat(properties.getPartitionCount()).isNull();
bind(Map.of("spring.cloud.stream.pulsar.binder.partition-count", "5150"));
assertThat(properties.getPartitionCount()).isEqualTo(5150);
assertThat(new PulsarBinderConfigurationProperties().getPartitionCount()).isNull();
var binderConfigProps = bindInputPropsToBinderConfigProps(
Map.of("spring.cloud.stream.pulsar.binder.partition-count", "5150"));
assertThat(binderConfigProps.getPartitionCount()).isEqualTo(5150);
}
@Test
void producerProperties() {
// Only spot check a few values (PulsarPropertiesTests does the heavy lifting)
Map<String, String> props = new HashMap<>();
props.put("spring.cloud.stream.pulsar.binder.producer.topic-name", "my-topic");
props.put("spring.cloud.stream.pulsar.binder.producer.send-timeout", "2s");
props.put("spring.cloud.stream.pulsar.binder.producer.max-pending-messages", "3");
props.put("spring.cloud.stream.pulsar.binder.producer.producer-access-mode", "exclusive");
props.put("spring.cloud.stream.pulsar.binder.producer.properties[my-prop]", "my-prop-value");
bind(props);
Map<String, Object> producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties.getProducer());
// Only spot check a few values (ProducerConfigPropertiesTests does the heavy
// lifting)
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.cloud.stream.pulsar.binder.producer.topic-name", "my-topic");
inputProps.put("spring.cloud.stream.pulsar.binder.producer.send-timeout", "2s");
inputProps.put("spring.cloud.stream.pulsar.binder.producer.max-pending-messages", "3");
inputProps.put("spring.cloud.stream.pulsar.binder.producer.access-mode", "exclusive");
var binderConfigProps = bindInputPropsToBinderConfigProps(inputProps);
var producerProps = binderConfigProps.getProducer().toAllProducerPropertiesMap();
// Verify that the props can be loaded in a ProducerBuilder
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
new ProducerConfigurationData(), ProducerConfigurationData.class));
// @formatter:off
assertThat(producerProps)
.containsEntry("topicName", "my-topic")
.containsEntry("sendTimeoutMs", 2_000)
.containsEntry("maxPendingMessages", 3)
.containsEntry("accessMode", ProducerAccessMode.Exclusive)
.hasEntrySatisfying("properties", properties ->
assertThat(properties)
.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
.containsEntry("my-prop", "my-prop-value"));
// @formatter:on
assertThat(producerProps).containsEntry("topicName", "my-topic").containsEntry("sendTimeoutMs", 2_000)
.containsEntry("maxPendingMessages", 3).containsEntry("accessMode", ProducerAccessMode.Exclusive);
}
@Test
void consumerProperties() {
// Only spot check a few values (PulsarPropertiesTests does the heavy lifting)
Map<String, String> props = new HashMap<>();
props.put("spring.cloud.stream.pulsar.binder.consumer.topics[0]", "my-topic");
props.put("spring.cloud.stream.pulsar.binder.consumer.subscription-properties[my-sub-prop]",
"my-sub-prop-value");
props.put("spring.cloud.stream.pulsar.binder.consumer.subscription-mode", "nondurable");
props.put("spring.cloud.stream.pulsar.binder.consumer.receiver-queue-size", "1");
bind(props);
Map<String, Object> consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getConsumer());
// Only spot check a few values (ConsumerConfigPropertiesTests does the heavy
// lifting)
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.cloud.stream.pulsar.binder.consumer.topics[0]", "my-topic");
inputProps.put("spring.cloud.stream.pulsar.binder.consumer.subscription.mode", "nondurable");
inputProps.put("spring.cloud.stream.pulsar.binder.consumer.receiver-queue-size", "1");
var binderConfigProps = bindInputPropsToBinderConfigProps(inputProps);
var consumerProps = binderConfigProps.getConsumer().toAllConsumerPropertiesMap();
// Verify that the props can be loaded in a ConsumerBuilder
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps,
new ConsumerConfigurationData<>(), ConsumerConfigurationData.class));
// @formatter:off
assertThat(consumerProps)
.hasEntrySatisfying("topicNames",
topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.collection(String.class))
.containsExactly("my-topic"))
.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("receiverQueueSize", 1);
// @formatter:on
.containsEntry("subscriptionMode", SubscriptionMode.NonDurable).containsEntry("receiverQueueSize", 1);
}
private PulsarBinderConfigurationProperties bindInputPropsToBinderConfigProps(Map<String, String> inputProps) {
return new Binder(new MapConfigurationPropertySource(inputProps)).bind("spring.cloud.stream.pulsar.binder",
Bindable.ofInstance(new PulsarBinderConfigurationProperties())).get();
}
}

View File

@@ -26,7 +26,6 @@ import java.util.function.Consumer;
import java.util.function.Supplier;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.schema.JSONSchema;
@@ -38,6 +37,8 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.WebApplicationType;
@@ -47,11 +48,9 @@ import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.pulsar.autoconfigure.PulsarProperties;
import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
@@ -60,7 +59,6 @@ import org.springframework.pulsar.core.ProducerBuilderCustomizer;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.support.header.PulsarHeaderMapper;
import org.springframework.pulsar.support.header.ToStringPulsarHeaderMapper;
@@ -84,17 +82,18 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.producer.cache.enabled=false",
"--spring.cloud.function.definition=textSupplier;textLogger",
"--spring.cloud.stream.bindings.textLogger-in-0.destination=textSupplier-out-0",
"--spring.pulsar.producer.producer-name=textSupplierProducer-fromBase",
"--spring.cloud.stream.pulsar.binder.producer.producer-name=textSupplierProducer-fromBinder",
"--spring.cloud.stream.pulsar.bindings.textSupplier-out-0.producer.producer-name=textSupplierProducer-fromBinding",
"--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.pulsar.producer.block-if-queue-full=true",
"--spring.cloud.stream.pulsar.binder.consumer.subscription-name=textLoggerSub-fromBinder",
"--spring.cloud.stream.pulsar.binder.consumer.consumer-name=textLogger-fromBinder",
"--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.consumer-name=textLogger-fromBinding")) {
"--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",
"--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.name=textLogger-fromBinding")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: test-basic-scenario"));
@@ -122,10 +121,10 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=textSupplier;textLogger",
"--spring.cloud.stream.bindings.textLogger-in-0.destination=textSupplier-out-0",
"--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.subscription-name=pbit-text-sub1")) {
"--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.subscription.name=pbit-text-sub1")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: test-basic-scenario"));
}
@@ -137,11 +136,11 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=piSupplier;piLogger",
"--spring.cloud.stream.bindings.piSupplier-out-0.destination=pi-stream",
"--spring.cloud.stream.bindings.piLogger-in-0.destination=pi-stream",
"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=pbit-float-sub1")) {
"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription.name=pbit-float-sub1")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: 3.14"));
}
@@ -158,14 +157,14 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=piSupplier;piLogger",
"--spring.cloud.stream.bindings.piLogger-in-0.destination=piSupplier-out-0",
"--spring.cloud.stream.bindings.piSupplier-out-0.producer.use-native-encoding=true",
"--spring.cloud.stream.pulsar.bindings.piSupplier-out-0.producer.schema-type=FLOAT",
"--spring.cloud.stream.bindings.piLogger-in-0.consumer.use-native-decoding=true",
"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.schema-type=FLOAT",
"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=pbit-float-sub2")) {
"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription.name=pbit-float-sub2")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: 3.14"));
}
@@ -177,7 +176,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=fooSupplier;fooLogger",
"--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-1",
"--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-1",
@@ -188,7 +187,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
"--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true",
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.schema-type=JSON",
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(),
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub1")) {
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription.name=pbit-foo-sub1")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: Foo[value=5150]"));
}
@@ -200,7 +199,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=fooSupplier;fooLogger",
"--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-2",
"--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-2",
@@ -209,7 +208,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+ Foo.class.getName(),
"--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true",
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(),
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub2")) {
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription.name=pbit-foo-sub2")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: Foo[value=5150]"));
}
@@ -221,7 +220,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=userSupplier;userLogger",
"--spring.cloud.stream.bindings.userSupplier-out-0.destination=user-stream-1",
"--spring.cloud.stream.bindings.userLogger-in-0.destination=user-stream-1",
@@ -233,7 +232,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.schema-type=AVRO",
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type="
+ User.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-user-sub1")) {
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription.name=pbit-user-sub1")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}"));
}
@@ -245,7 +244,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=userSupplier;userLogger",
"--spring.cloud.stream.bindings.userSupplier-out-0.destination=user-stream-2",
"--spring.cloud.stream.bindings.userLogger-in-0.destination=user-stream-2",
@@ -255,7 +254,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
"--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true",
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type="
+ User.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-user-sub2",
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription.name=pbit-user-sub2",
"--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()),
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
@@ -269,7 +268,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=userSupplier;userLogger",
"--spring.cloud.stream.bindings.userSupplier-out-0.destination=user-stream-3",
"--spring.cloud.stream.bindings.userLogger-in-0.destination=user-stream-3",
@@ -279,7 +278,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
"--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true",
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type="
+ User.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-user-sub3")) {
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription.name=pbit-user-sub3")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}"));
}
@@ -291,7 +290,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=userSupplier;userLogger",
"--spring.cloud.stream.bindings.userSupplier-out-0.destination=kv-stream-1",
"--spring.cloud.stream.bindings.userLogger-in-0.destination=kv-stream-1",
@@ -307,7 +306,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+ User.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-key-type="
+ String.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub1",
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription.name=pbit-kv-sub1",
"--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()),
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
@@ -321,7 +320,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=userSupplier;userLogger",
"--spring.cloud.stream.bindings.userSupplier-out-0.destination=kv-stream-2",
"--spring.cloud.stream.bindings.userLogger-in-0.destination=kv-stream-2",
@@ -335,7 +334,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+ User.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-key-type="
+ String.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub2",
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription.name=pbit-kv-sub2",
"--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()),
"--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
@@ -349,7 +348,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=userSupplier;userLogger",
"--spring.cloud.stream.bindings.userSupplier-out-0.destination=kv-stream-3",
"--spring.cloud.stream.bindings.userLogger-in-0.destination=kv-stream-3",
@@ -365,7 +364,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+ User.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-key-type="
+ String.class.getName(),
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub3")) {
"--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription.name=pbit-kv-sub3")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}"));
}
@@ -377,7 +376,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=fooSupplier;fooLogger",
"--spring.cloud.stream.bindings.fooSupplier-out-0.destination=kv-stream-4",
"--spring.cloud.stream.bindings.fooLogger-in-0.destination=kv-stream-4",
@@ -390,7 +389,7 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(),
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-key-type="
+ String.class.getName(),
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-kv-sub4")) {
"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription.name=pbit-kv-sub4")) {
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: 5150->Foo[value=5150]"));
}
@@ -407,11 +406,11 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=springMessageSupplier;springMessageLogger",
"--spring.cloud.stream.bindings.springMessageSupplier-out-0.destination=cmh-1",
"--spring.cloud.stream.bindings.springMessageLogger-in-0.destination=cmh-1",
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription-name=pbit-cmh1-sub1")) {
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription.name=pbit-cmh1-sub1")) {
// Wait for a few of the messages to flow through (check for index = 5)
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)).until(
() -> output.toString().contains("Hello binder: test-headers-msg-5 w/ custom-id: 5150-5"));
@@ -424,11 +423,11 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=springMessageSupplier;springMessageLogger",
"--spring.cloud.stream.bindings.springMessageSupplier-out-0.destination=cmh-2",
"--spring.cloud.stream.bindings.springMessageLogger-in-0.destination=cmh-2",
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription-name=pbit-cmh2-sub1")) {
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription.name=pbit-cmh2-sub1")) {
// Wait for a few of the messages to flow through (check for index = 5)
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)).until(() -> output.toString()
.contains("Hello binder: test-headers-msg-5 w/ custom-id: FooHeader[value=5150-5]"));
@@ -441,12 +440,12 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=springMessageSupplier;springMessageLogger",
"--spring.cloud.stream.bindings.springMessageSupplier-out-0.destination=cmh-3",
"--spring.cloud.stream.bindings.springMessageSupplier-out-0.producer.header-mode=none",
"--spring.cloud.stream.bindings.springMessageLogger-in-0.destination=cmh-3",
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription-name=pbit-cmh3-sub1")) {
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription.name=pbit-cmh3-sub1")) {
// Wait for a few of the messages to flow through (check for index = 5)
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: test-headers-msg-5 w/ custom-id: null"));
@@ -459,12 +458,12 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=springMessageSupplier;springMessageLogger",
"--spring.cloud.stream.bindings.springMessageSupplier-out-0.destination=cmh-4",
"--spring.cloud.stream.bindings.springMessageLogger-in-0.destination=cmh-4",
"--spring.cloud.stream.bindings.springMessageLogger-in-0.consumer.header-mode=none",
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription-name=pbit-cmh4-sub1")) {
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription.name=pbit-cmh4-sub1")) {
// Wait for a few of the messages to flow through (check for index = 5)
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
.until(() -> output.toString().contains("Hello binder: test-headers-msg-5 w/ custom-id: null"));
@@ -477,11 +476,11 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=springMessageSupplier;springMessageLogger",
"--spring.cloud.stream.bindings.springMessageSupplier-out-0.destination=cmh-5",
"--spring.cloud.stream.bindings.springMessageLogger-in-0.destination=cmh-5",
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription-name=pbit-cmh5-sub1")) {
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription.name=pbit-cmh5-sub1")) {
// Wait for a few of the messages to flow through (check for index = 5)
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)).until(() -> output.toString()
.contains("Hello binder: test-headers-msg-5 w/ custom-id: tsh->tph->FooHeader[value=5150-5]"));
@@ -638,66 +637,109 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
@Import(PrimitiveTextConfig.class)
static class BinderAndBindingPropsTestConfig {
@SuppressWarnings("unchecked")
@Bean
public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient,
PulsarProperties pulsarProperties, TopicResolver topicResolver) {
var customizer = (ProducerBuilderCustomizer<String>) pulsarProperties.getProducer()
.toProducerBuilderCustomizer();
return new TrackingProducerFactory(pulsarClient, pulsarProperties.getProducer().getTopicName(), customizer,
topicResolver);
TrackingProducerFactoryBeanPostProcessor trackingProducerFactory() {
return new TrackingProducerFactoryBeanPostProcessor();
}
@SuppressWarnings("unchecked")
@Bean
public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
PulsarProperties pulsarProperties) {
var customizer = (ConsumerBuilderCustomizer<String>) pulsarProperties.getConsumer()
.toConsumerBuilderCustomizer();
return new TrackingConsumerFactory(pulsarClient, customizer);
TrackingConsumerFactoryBeanPostProcessor trackingConsumerFactory() {
return new TrackingConsumerFactoryBeanPostProcessor();
}
}
static class TrackingProducerFactory extends DefaultPulsarProducerFactory<String> {
static class TrackingProducerFactoryBeanPostProcessor implements BeanPostProcessor {
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof DefaultPulsarProducerFactory defaultFactory) {
return new TrackingProducerFactory(defaultFactory);
}
return bean;
}
}
static class TrackingProducerFactory implements PulsarProducerFactory<String> {
private final DefaultPulsarProducerFactory<String> trackedProducerFactory;
List<Producer<String>> producersCreated = new ArrayList<>();
TrackingProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
ProducerBuilderCustomizer<String> defaultConfigCustomizer, TopicResolver topicResolver) {
super(pulsarClient, defaultTopic, defaultConfigCustomizer, topicResolver);
TrackingProducerFactory(DefaultPulsarProducerFactory<String> trackedProducerFactory) {
this.trackedProducerFactory = trackedProducerFactory;
}
@Override
protected Producer<String> doCreateProducer(Schema<String> schema, @Nullable String topic,
@Nullable Collection<String> encryptionKeys,
@Nullable List<ProducerBuilderCustomizer<String>> producerBuilderCustomizers)
throws PulsarClientException {
Producer<String> producer = super.doCreateProducer(schema, topic, encryptionKeys,
producerBuilderCustomizers);
producersCreated.add(producer);
public Producer<String> createProducer(Schema<String> schema, String topic) throws PulsarClientException {
var producer = this.trackedProducerFactory.createProducer(schema, topic);
this.producersCreated.add(producer);
return producer;
}
@Override
public Producer<String> createProducer(Schema<String> schema, String topic,
ProducerBuilderCustomizer<String> customizer) throws PulsarClientException {
var producer = this.trackedProducerFactory.createProducer(schema, topic, customizer);
this.producersCreated.add(producer);
return producer;
}
@Override
public Producer<String> createProducer(Schema<String> schema, String topic, Collection<String> encryptionKeys,
List<ProducerBuilderCustomizer<String>> producerBuilderCustomizers) throws PulsarClientException {
var producer = this.trackedProducerFactory.createProducer(schema, topic, encryptionKeys,
producerBuilderCustomizers);
this.producersCreated.add(producer);
return producer;
}
@Override
public String getDefaultTopic() {
return this.trackedProducerFactory.getDefaultTopic();
}
}
static class TrackingConsumerFactory extends DefaultPulsarConsumerFactory<String> {
static class TrackingConsumerFactoryBeanPostProcessor implements BeanPostProcessor {
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof DefaultPulsarConsumerFactory defaultFactory) {
return new TrackingConsumerFactory(defaultFactory);
}
return bean;
}
}
static class TrackingConsumerFactory implements PulsarConsumerFactory<String> {
private final DefaultPulsarConsumerFactory<String> trackedConsumerFactory;
List<org.apache.pulsar.client.api.Consumer<String>> consumersCreated = new ArrayList<>();
TrackingConsumerFactory(PulsarClient pulsarClient, ConsumerBuilderCustomizer<String> defaultConsumerConfig) {
super(pulsarClient, defaultConsumerConfig);
TrackingConsumerFactory(DefaultPulsarConsumerFactory<String> trackedConsumerFactory) {
this.trackedConsumerFactory = trackedConsumerFactory;
}
@Override
public org.apache.pulsar.client.api.Consumer<String> createConsumer(Schema<String> schema,
@Nullable Collection<String> topics, @Nullable String subscriptionName,
@Nullable Map<String, String> metadataProperties,
@Nullable List<ConsumerBuilderCustomizer<String>> consumerBuilderCustomizers)
Collection<String> topics, String subscriptionName, ConsumerBuilderCustomizer<String> customizer)
throws PulsarClientException {
org.apache.pulsar.client.api.Consumer<String> consumer = super.createConsumer(schema, topics,
subscriptionName, metadataProperties, consumerBuilderCustomizers);
consumersCreated.add(consumer);
var consumer = this.trackedConsumerFactory.createConsumer(schema, topics, subscriptionName, customizer);
this.consumersCreated.add(consumer);
return consumer;
}
@Override
public org.apache.pulsar.client.api.Consumer<String> createConsumer(Schema<String> schema,
Collection<String> topics, String subscriptionName, Map<String, String> metadataProperties,
List<ConsumerBuilderCustomizer<String>> consumerBuilderCustomizers) throws PulsarClientException {
var consumer = this.trackedConsumerFactory.createConsumer(schema, topics, subscriptionName,
metadataProperties, consumerBuilderCustomizers);
this.consumersCreated.add(consumer);
return consumer;
}

View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.stream.binder.pulsar;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -104,7 +105,8 @@ public class PulsarBinderTests extends
var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
var pulsarTemplate = new PulsarTemplate<>(producerFactory);
var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
(consumerBuilder -> consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)));
List.of((consumerBuilder) -> consumerBuilder
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)));
if (this.binder == null) {
this.binder = new PulsarTestBinder(provisioner, pulsarTemplate, consumerFactory, configProps,
new DefaultSchemaResolver(), JsonPulsarHeaderMapper.builder().build());
@@ -155,8 +157,7 @@ public class PulsarBinderTests extends
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.bar", null, moduleInputChannel,
consumerProperties);
Message<?> message = MessageBuilder
.withPayload("foo".getBytes(StandardCharsets.UTF_8)).build();
Message<?> message = MessageBuilder.withPayload("foo".getBytes(StandardCharsets.UTF_8)).build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();

View File

@@ -17,42 +17,24 @@
package org.springframework.cloud.stream.binder.pulsar;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Stream;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.HashingScheme;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.ProducerAccessMode;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Mockito;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.ConfigurationPropertySource;
import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
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;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.junit.jupiter.params.provider.Arguments.arguments;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -71,8 +53,8 @@ public class PulsarBinderUtilsTests {
@Test
void respectsValueWhenSetAsProperty() {
var consumerDestination = mock(ConsumerDestination.class);
var pulsarConsumerProperties = mock(PulsarConsumerProperties.class);
when(pulsarConsumerProperties.getSubscriptionName()).thenReturn("my-sub");
var pulsarConsumerProperties = mock(PulsarConsumerProperties.class, Mockito.RETURNS_DEEP_STUBS);
when(pulsarConsumerProperties.getSubscription().getName()).thenReturn("my-sub");
assertThat(PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination))
.isEqualTo("my-sub");
}
@@ -80,8 +62,8 @@ public class PulsarBinderUtilsTests {
@Test
void generatesValueWhenNotSetAsProperty() {
var consumerDestination = mock(ConsumerDestination.class);
var pulsarConsumerProperties = mock(PulsarConsumerProperties.class);
when(pulsarConsumerProperties.getSubscriptionName()).thenReturn(null);
var pulsarConsumerProperties = mock(PulsarConsumerProperties.class, Mockito.RETURNS_DEEP_STUBS);
when(pulsarConsumerProperties.getSubscription().getName()).thenReturn(null);
when(consumerDestination.getName()).thenReturn("my-topic");
assertThat(PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination))
.startsWith("my-topic-anon-subscription-");
@@ -154,169 +136,162 @@ public class PulsarBinderUtilsTests {
}
@Nested
class ConvertedPropertiesTests {
class MergeProducerPropertiesTests {
private final ProducerConfigProperties properties = new ProducerConfigProperties();
private static final Consumer<ProducerConfigProperties> SET_NO_PROPS = (__) -> {
};
private void bind(Map<String, String> map) {
ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
new Binder(source).bind("spring.pulsar.producer", Bindable.ofInstance(this.properties));
@Test
void noPropsSpecified() {
doMergeProducerPropertiesTest(SET_NO_PROPS, SET_NO_PROPS, Collections.emptyMap());
}
@Test
void producerPropertiesToMap() {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.producer.topic-name", "my-topic");
props.put("spring.pulsar.producer.producer-name", "my-producer");
props.put("spring.pulsar.producer.send-timeout", "2s");
props.put("spring.pulsar.producer.block-if-queue-full", "true");
props.put("spring.pulsar.producer.max-pending-messages", "3");
props.put("spring.pulsar.producer.max-pending-messages-across-partitions", "4");
props.put("spring.pulsar.producer.message-routing-mode", "custompartition");
props.put("spring.pulsar.producer.hashing-scheme", "murmur3_32hash");
props.put("spring.pulsar.producer.crypto-failure-action", "send");
props.put("spring.pulsar.producer.batching-max-publish-delay", "5s");
props.put("spring.pulsar.producer.batching-partition-switch-frequency-by-publish-delay", "6");
props.put("spring.pulsar.producer.batching-max-messages", "7");
props.put("spring.pulsar.producer.batching-max-bytes", "8");
props.put("spring.pulsar.producer.batching-enabled", "false");
props.put("spring.pulsar.producer.chunking-enabled", "true");
props.put("spring.pulsar.producer.encryption-keys[0]", "my-key");
props.put("spring.pulsar.producer.compression-type", "lz4");
props.put("spring.pulsar.producer.initial-sequence-id", "9");
props.put("spring.pulsar.producer.producer-access-mode", "exclusive");
props.put("spring.pulsar.producer.lazy-start=partitioned-producers", "true");
props.put("spring.pulsar.producer.properties[my-prop]", "my-prop-value");
void basePropSpecifiedAtBinderLevelOnly() {
doMergeProducerPropertiesTest((binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Exclusive),
SET_NO_PROPS, Map.of("accessMode", ProducerAccessMode.Exclusive));
}
bind(props);
Map<String, Object> producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties);
@Test
void basePropSpecifiedAtBindingLevelOnly() {
doMergeProducerPropertiesTest(SET_NO_PROPS,
(bindingProps) -> bindingProps.setAccessMode(ProducerAccessMode.Exclusive),
Map.of("accessMode", ProducerAccessMode.Exclusive));
}
// Verify that the props can be loaded in a ProducerBuilder
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
new ProducerConfigurationData(), ProducerConfigurationData.class));
@Test
void basePropSpecifiedAtBinderAndBindingLevel() {
doMergeProducerPropertiesTest(
(binderProps) -> binderProps.setAccessMode(ProducerAccessMode.ExclusiveWithFencing),
(bindingProps) -> bindingProps.setAccessMode(ProducerAccessMode.Exclusive),
Map.of("accessMode", ProducerAccessMode.Exclusive));
}
assertThat(producerProps).containsEntry("topicName", "my-topic")
.containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000)
.containsEntry("blockIfQueueFull", true).containsEntry("maxPendingMessages", 3)
.containsEntry("maxPendingMessagesAcrossPartitions", 4)
.containsEntry("messageRoutingMode", MessageRoutingMode.CustomPartition)
.containsEntry("hashingScheme", HashingScheme.Murmur3_32Hash)
.containsEntry("cryptoFailureAction", ProducerCryptoFailureAction.SEND)
.containsEntry("batchingMaxPublishDelayMicros", 5_000_000L)
.containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6)
.containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8)
.containsEntry("batchingEnabled", false).containsEntry("chunkingEnabled", true)
.hasEntrySatisfying("encryptionKeys",
keys -> assertThat(keys).asInstanceOf(InstanceOfAssertFactories.collection(String.class))
.containsExactly("my-key"))
.containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L)
.containsEntry("accessMode", ProducerAccessMode.Exclusive)
.containsEntry("lazyStartPartitionedProducers", true).hasEntrySatisfying("properties",
properties -> assertThat(properties)
.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
.containsEntry("my-prop", "my-prop-value"));
@Test
void basePropSpecifiedWithSameValueAsDefault() {
doMergeProducerPropertiesTest((binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Shared),
(bindingProps) -> bindingProps.setAccessMode(ProducerAccessMode.Shared), Collections.emptyMap());
}
@Test
void extPropSpecifiedAtBinderLevelOnly() {
doMergeProducerPropertiesTest((binderProps) -> binderProps.setMaxPendingMessages(1200), SET_NO_PROPS,
Map.of("maxPendingMessages", 1200));
}
@Test
void extPropSpecifiedAtBindingLevelOnly() {
doMergeProducerPropertiesTest(SET_NO_PROPS, (bindingProps) -> bindingProps.setMaxPendingMessages(1200),
Map.of("maxPendingMessages", 1200));
}
@Test
void extPropSpecifiedAtBinderAndBindingLevel() {
doMergeProducerPropertiesTest((binderProps) -> binderProps.setMaxPendingMessages(1100),
(bindingProps) -> bindingProps.setMaxPendingMessages(1200), Map.of("maxPendingMessages", 1200));
}
@Test
void extPropSpecifiedWithSameValueAsDefault() {
doMergeProducerPropertiesTest((binderProps) -> binderProps.setMaxPendingMessages(1000),
(bindingProps) -> bindingProps.setMaxPendingMessages(1000), Collections.emptyMap());
}
@Test
void baseAndExtPropsAreCombined() {
doMergeProducerPropertiesTest((binderProps) -> binderProps.setAccessMode(ProducerAccessMode.Exclusive),
(bindingProps) -> bindingProps.setMaxPendingMessages(1200),
Map.of("accessMode", ProducerAccessMode.Exclusive, "maxPendingMessages", 1200));
}
void doMergeProducerPropertiesTest(Consumer<ProducerConfigProperties> binderPropsCustomizer,
Consumer<ProducerConfigProperties> bindingPropsCustomizer, Map<String, Object> expectedProps) {
var binderProducerProps = new ProducerConfigProperties();
binderPropsCustomizer.accept(binderProducerProps);
var bindingProducerProps = new ProducerConfigProperties();
bindingPropsCustomizer.accept(bindingProducerProps);
var mergedProps = PulsarBinderUtils.mergeModifiedProducerProperties(binderProducerProps,
bindingProducerProps);
assertThat(mergedProps).containsExactlyInAnyOrderEntriesOf(expectedProps);
}
}
@Nested
class ConvertedConsumerPropertiesTests {
class MergeConsumerPropertiesTests {
private final ConsumerConfigProperties properties = new ConsumerConfigProperties();
private static final Consumer<ConsumerConfigProperties> SET_NO_PROPS = (__) -> {
};
private void bind(Map<String, String> map) {
ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
new Binder(source).bind("spring.pulsar.consumer", Bindable.ofInstance(this.properties));
@Test
void noPropsSpecified() {
doMergeConsumerPropertiesTest(SET_NO_PROPS, SET_NO_PROPS, Collections.emptyMap());
}
@Test
void consumerPropertiesToMap() {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.consumer.topics[0]", "my-topic");
props.put("spring.pulsar.consumer.topics-pattern", "my-pattern");
props.put("spring.pulsar.consumer.subscription-name", "my-subscription");
props.put("spring.pulsar.consumer.subscription-type", "shared");
props.put("spring.pulsar.consumer.subscription-properties[my-sub-prop]", "my-sub-prop-value");
props.put("spring.pulsar.consumer.subscription-mode", "nondurable");
props.put("spring.pulsar.consumer.receiver-queue-size", "1");
props.put("spring.pulsar.consumer.acknowledgements-group-time", "2s");
props.put("spring.pulsar.consumer.negative-ack-redelivery-delay", "3s");
props.put("spring.pulsar.consumer.max-total-receiver-queue-size-across-partitions", "5");
props.put("spring.pulsar.consumer.consumer-name", "my-consumer");
props.put("spring.pulsar.consumer.ack-timeout", "6s");
props.put("spring.pulsar.consumer.tick-duration", "7s");
props.put("spring.pulsar.consumer.priority-level", "8");
props.put("spring.pulsar.consumer.crypto-failure-action", "discard");
props.put("spring.pulsar.consumer.properties[my-prop]", "my-prop-value");
props.put("spring.pulsar.consumer.read-compacted", "true");
props.put("spring.pulsar.consumer.subscription-initial-position", "earliest");
props.put("spring.pulsar.consumer.pattern-auto-discovery-period", "9");
props.put("spring.pulsar.consumer.regex-subscription-mode", "all-topics");
props.put("spring.pulsar.consumer.dead-letter-policy.max-redeliver-count", "4");
props.put("spring.pulsar.consumer.dead-letter-policy.retry-letter-topic", "my-retry-topic");
props.put("spring.pulsar.consumer.dead-letter-policy.dead-letter-topic", "my-dlt-topic");
props.put("spring.pulsar.consumer.dead-letter-policy.initial-subscription-name", "my-initial-subscription");
props.put("spring.pulsar.consumer.retry-enable", "true");
props.put("spring.pulsar.consumer.auto-update-partitions", "false");
props.put("spring.pulsar.consumer.auto-update-partitions-interval", "10s");
props.put("spring.pulsar.consumer.replicate-subscription-state", "true");
props.put("spring.pulsar.consumer.reset-include-head", "true");
props.put("spring.pulsar.consumer.batch-index-ack-enabled", "true");
props.put("spring.pulsar.consumer.ack-receipt-enabled", "true");
props.put("spring.pulsar.consumer.pool-messages", "true");
props.put("spring.pulsar.consumer.start-paused", "true");
props.put("spring.pulsar.consumer.auto-ack-oldest-chunked-message-on-queue-full", "false");
props.put("spring.pulsar.consumer.max-pending-chunked-message", "11");
props.put("spring.pulsar.consumer.expire-time-of-incomplete-chunked-message", "12s");
void basePropSpecifiedAtBinderLevelOnly() {
doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(1000), SET_NO_PROPS,
Map.of("priorityLevel", 1000));
}
bind(props);
Map<String, Object> consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties);
@Test
void basePropSpecifiedAtBindingLevelOnly() {
doMergeConsumerPropertiesTest(SET_NO_PROPS, (bindingProps) -> bindingProps.setPriorityLevel(1000),
Map.of("priorityLevel", 1000));
}
// Verify that the props can be loaded in a ConsumerBuilder
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps,
new ConsumerConfigurationData<>(), ConsumerConfigurationData.class));
@Test
void basePropSpecifiedAtBinderAndBindingLevel() {
doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(2000),
(bindingProps) -> bindingProps.setPriorityLevel(1000), Map.of("priorityLevel", 1000));
}
assertThat(consumerProps)
.hasEntrySatisfying("topicNames",
topics -> assertThat(topics)
.asInstanceOf(InstanceOfAssertFactories.collection(String.class))
.containsExactly("my-topic"))
.hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern"))
.containsEntry("subscriptionName", "my-subscription")
.containsEntry("subscriptionType", SubscriptionType.Shared)
.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("receiverQueueSize", 1).containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L)
.containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L)
.containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5)
.containsEntry("consumerName", "my-consumer").containsEntry("ackTimeoutMillis", 6_000L)
.containsEntry("tickDurationMillis", 7_000L).containsEntry("priorityLevel", 8)
.containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD)
.hasEntrySatisfying("properties",
properties -> assertThat(properties)
.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
.containsEntry("my-prop", "my-prop-value"))
.containsEntry("readCompacted", true)
.containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest)
.containsEntry("patternAutoDiscoveryPeriod", 9)
.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");
}).containsEntry("retryEnable", true).containsEntry("autoUpdatePartitions", false)
.containsEntry("autoUpdatePartitionsIntervalSeconds", 10L)
.containsEntry("replicateSubscriptionState", true).containsEntry("resetIncludeHead", true)
.containsEntry("batchIndexAckEnabled", true).containsEntry("ackReceiptEnabled", true)
.containsEntry("poolMessages", true).containsEntry("startPaused", true)
.containsEntry("autoAckOldestChunkedMessageOnQueueFull", false)
.containsEntry("maxPendingChunkedMessage", 11)
.containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L);
@Test
void basePropSpecifiedWithSameValueAsDefault() {
doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(0),
(bindingProps) -> bindingProps.setPriorityLevel(0), Collections.emptyMap());
}
@Test
void extPropSpecifiedAtBinderLevelOnly() {
doMergeConsumerPropertiesTest((binderProps) -> binderProps.setReceiverQueueSize(1200), SET_NO_PROPS,
Map.of("receiverQueueSize", 1200));
}
@Test
void extPropSpecifiedAtBindingLevelOnly() {
doMergeConsumerPropertiesTest(SET_NO_PROPS, (bindingProps) -> bindingProps.setReceiverQueueSize(1200),
Map.of("receiverQueueSize", 1200));
}
@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());
}
@Test
void baseAndExtPropsAreCombined() {
doMergeConsumerPropertiesTest((binderProps) -> binderProps.setPriorityLevel(1000),
(bindingProps) -> bindingProps.setReceiverQueueSize(1200),
Map.of("priorityLevel", 1000, "receiverQueueSize", 1200));
}
void doMergeConsumerPropertiesTest(Consumer<ConsumerConfigProperties> binderPropsCustomizer,
Consumer<ConsumerConfigProperties> bindingPropsCustomizer, Map<String, Object> expectedProps) {
var binderConsumerProps = new ConsumerConfigProperties();
binderPropsCustomizer.accept(binderConsumerProps);
var bindingConsumerProps = new ConsumerConfigProperties();
bindingPropsCustomizer.accept(bindingConsumerProps);
var mergedProps = PulsarBinderUtils.mergeModifiedConsumerProperties(binderConsumerProps,
bindingConsumerProps);
assertThat(mergedProps).containsExactlyInAnyOrderEntriesOf(expectedProps);
}
}

View File

@@ -30,7 +30,6 @@ import org.junit.jupiter.api.Test;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.ConfigurationPropertySource;
import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
import org.springframework.cloud.stream.binder.pulsar.properties.PulsarExtendedBindingProperties;
import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -45,96 +44,68 @@ import static org.assertj.core.api.Assertions.assertThatNoException;
*/
public class PulsarExtendedBindingPropertiesTests {
private final PulsarExtendedBindingProperties properties = new PulsarExtendedBindingProperties();
private void bind(Map<String, String> map) {
ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
new Binder(source).bind("spring.cloud.stream.pulsar", Bindable.ofInstance(this.properties));
}
@Test
void producerProperties() {
// Only spot check a few values (PulsarPropertiesTests does the heavy lifting)
Map<String, String> props = new HashMap<>();
props.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.topic-name", "my-topic");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.send-timeout", "2s");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.max-pending-messages", "3");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.producer-access-mode", "exclusive");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.properties[my-prop]", "my-prop-value");
// Only spot check a few values (ProducerConfigPropertiesTests does the heavy
// lifting)
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.topic-name", "my-topic");
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.send-timeout", "2s");
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.max-pending-messages", "3");
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.producer.access-mode", "exclusive");
var bindingConfigProps = bindInputPropsToBindingConfigProps(inputProps);
assertThat(bindingConfigProps.getBindings()).containsOnlyKeys("my-foo");
var producerProps = bindingConfigProps.getExtendedProducerProperties("my-foo").toAllProducerPropertiesMap();
bind(props);
assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
Map<String, Object> producerProps = PulsarBinderUtils
.convertProducerPropertiesToMap(properties.getExtendedProducerProperties("my-foo"));
// Verify that the props can be loaded in a ProducerBuilder
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
new ProducerConfigurationData(), ProducerConfigurationData.class));
// @formatter:off
assertThat(producerProps)
.containsEntry("topicName", "my-topic")
.containsEntry("sendTimeoutMs", 2_000)
.containsEntry("maxPendingMessages", 3)
.containsEntry("accessMode", ProducerAccessMode.Exclusive)
.hasEntrySatisfying("properties", properties ->
assertThat(properties)
.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
.containsEntry("my-prop", "my-prop-value"));
// @formatter:on
assertThat(producerProps).containsEntry("topicName", "my-topic").containsEntry("sendTimeoutMs", 2_000)
.containsEntry("maxPendingMessages", 3).containsEntry("accessMode", ProducerAccessMode.Exclusive);
}
@Test
void consumerProperties() {
// Only spot check a few values (PulsarPropertiesTests does the heavy lifting)
Map<String, String> props = new HashMap<>();
props.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.topics[0]", "my-topic");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.subscription-properties[my-sub-prop]",
"my-sub-prop-value");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.subscription-mode", "nondurable");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.receiver-queue-size", "1");
// Only spot check a few values (ConsumerConfigPropertiesTests does the heavy
// lifting)
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.topics[0]", "my-topic");
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.subscription.mode", "nondurable");
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.receiver-queue-size", "1");
var bindingConfigProps = bindInputPropsToBindingConfigProps(inputProps);
assertThat(bindingConfigProps.getBindings()).containsOnlyKeys("my-foo");
var consumerProps = bindingConfigProps.getExtendedConsumerProperties("my-foo").toAllConsumerPropertiesMap();
bind(props);
assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
Map<String, Object> consumerProps = PulsarBinderUtils
.convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo"));
// Verify that the props can be loaded in a ConsumerBuilder
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps,
new ConsumerConfigurationData<>(), ConsumerConfigurationData.class));
// @formatter:off
assertThat(consumerProps)
.hasEntrySatisfying("topicNames",
topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.collection(String.class))
.containsExactly("my-topic"))
.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("receiverQueueSize", 1);
// @formatter:on
.containsEntry("subscriptionMode", SubscriptionMode.NonDurable).containsEntry("receiverQueueSize", 1);
}
@Test
void extendedBindingsArePropagatedToContainerProperties() {
Map<String, String> props = new HashMap<>();
props.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.subscription-name", "my-foo-sbscription");
props.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.subscription-type", "Shared");
Map<String, String> inputProps = new HashMap<>();
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.subscription.name", "my-foo-sbscription");
inputProps.put("spring.cloud.stream.pulsar.bindings.my-foo.consumer.subscription.type", "Shared");
var bindingConfigProps = bindInputPropsToBindingConfigProps(inputProps);
var consumerProps = bindingConfigProps.getExtendedConsumerProperties("my-foo").toAllConsumerPropertiesMap();
bind(props);
var bindingConsumerProps = PulsarBinderUtils
.convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo"));
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.getPulsarConsumerProperties().putAll(bindingConsumerProps);
pulsarContainerProperties.getPulsarConsumerProperties().putAll(consumerProps);
assertThat(pulsarContainerProperties.getSubscriptionName()).isNull();
assertThat(pulsarContainerProperties.getSubscriptionType()).isEqualTo(SubscriptionType.Exclusive);
pulsarContainerProperties.updateContainerProperties();
assertThat(pulsarContainerProperties.getSubscriptionName()).isEqualTo("my-foo-sbscription");
assertThat(pulsarContainerProperties.getSubscriptionType()).isEqualTo(SubscriptionType.Shared);
}
private PulsarExtendedBindingProperties bindInputPropsToBindingConfigProps(Map<String, String> inputProps) {
return new Binder(new MapConfigurationPropertySource(inputProps))
.bind("spring.cloud.stream.pulsar", Bindable.ofInstance(new PulsarExtendedBindingProperties())).get();
}
}

View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.binder.pulsar;
import java.util.Locale;
import org.junit.jupiter.api.BeforeAll;
import org.testcontainers.containers.PulsarContainer;
import org.testcontainers.junit.jupiter.Testcontainers;
@@ -43,26 +41,11 @@ public interface PulsarTestContainerSupport {
}
static DockerImageName getPulsarImage() {
return isRunningOnMacM1() ? getMacM1PulsarImage() : getStandardPulsarImage();
return DockerImageName.parse("apachepulsar/pulsar:3.1.0");
}
static String getHttpServiceUrl() {
return PULSAR_CONTAINER.getHttpServiceUrl();
}
private static boolean isRunningOnMacM1() {
String osName = System.getProperty("os.name").toLowerCase(Locale.ENGLISH);
String osArchitecture = System.getProperty("os.arch").toLowerCase(Locale.ENGLISH);
return osName.contains("mac") && osArchitecture.equals("aarch64");
}
private static DockerImageName getStandardPulsarImage() {
return DockerImageName.parse("apachepulsar/pulsar:2.11.0");
}
private static DockerImageName getMacM1PulsarImage() {
return DockerImageName.parse("kezhenxu94/pulsar").asCompatibleSubstituteFor("apachepulsar/pulsar");
}
}