pulsarHeaders = this.delegate.toPulsarHeaders(springHeaders);
+ pulsarHeaders.remove(MessageHeaders.ID);
+ pulsarHeaders.remove(MessageHeaders.TIMESTAMP);
+ pulsarHeaders.remove(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT);
+ pulsarHeaders.remove(BinderHeaders.NATIVE_HEADERS_PRESENT);
+ return pulsarHeaders;
+ }
+
+ @Override
+ public MessageHeaders toSpringHeaders(Message pulsarMessage) {
+ var springHeaders = this.delegate.toSpringHeaders(pulsarMessage);
+ if (!springHeaders.isEmpty()) {
+ MessageHeaderAccessor mutableHeaders = new MessageHeaderAccessor();
+ mutableHeaders.copyHeaders(springHeaders);
+ mutableHeaders.setHeader(BinderHeaders.NATIVE_HEADERS_PRESENT, Boolean.TRUE);
+ springHeaders = mutableHeaders.getMessageHeaders();
+ }
+ return springHeaders;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java
new file mode 100644
index 000000000..51e6e18cd
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java
@@ -0,0 +1,104 @@
+/*
+ * 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 java.util.Objects;
+import java.util.UUID;
+
+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.util.StringUtils;
+
+
+/**
+ * Binder utility methods.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+final class PulsarBinderUtils {
+
+ private static final LogAccessor LOGGER = new LogAccessor(PulsarBinderUtils.class);
+
+ private static final String SUBSCRIPTION_NAME_FORMAT_STR = "%s-anon-subscription-%s";
+
+ private PulsarBinderUtils() {
+ }
+
+ /**
+ * Gets the subscription name to use for the binder.
+ * @param consumerProps the pulsar consumer props
+ * @param consumerDestination the destination being subscribed to
+ * @return the subscription name from the consumer properties or a generated name in
+ * the format {@link #SUBSCRIPTION_NAME_FORMAT_STR} when the name is not set on the
+ * consumer properties
+ */
+ static String subscriptionName(PulsarConsumerProperties consumerProps, ConsumerDestination consumerDestination) {
+ if (StringUtils.hasText(consumerProps.getSubscriptionName())) {
+ return consumerProps.getSubscriptionName();
+ }
+ return SUBSCRIPTION_NAME_FORMAT_STR.formatted(consumerDestination.getName(), UUID.randomUUID());
+ }
+
+ /**
+ * Merges properties defined at the binder and binding level (binding properties
+ * override binder properties).
+ *
+ * NOTE: Properties whose value is not different from the default value in the
+ * {@code baseProps} are not included in the merged result.
+ * @param baseProps the map of base level properties (eg. 'spring.pulsar.consumer.*')
+ * @param binderProps the map of binder level properties (eg.
+ * 'spring.cloud.stream.pulsar.binder.consumer.*')
+ * @param bindingProps the map of binding level properties (eg.
+ * 'spring.cloud.stream.pulsar.bindings.myBinding-in-0.consumer.*')
+ * @return map of merged binder and binding properties including only properties whose
+ * value has changed from the same property in the base properties
+ */
+ static Map mergePropertiesWithPrecedence(Map baseProps,
+ Map binderProps, Map bindingProps) {
+ Objects.requireNonNull(baseProps, "baseProps must be specified");
+ Objects.requireNonNull(binderProps, "binderProps must be specified");
+ Objects.requireNonNull(bindingProps, "bindingProps must be specified");
+
+ Map newOrModifiedBinderProps = extractNewOrModifiedProperties(binderProps, baseProps);
+ LOGGER.trace(() -> "New or modified binder props: %s".formatted(newOrModifiedBinderProps));
+
+ Map newOrModifiedBindingProps = extractNewOrModifiedProperties(bindingProps, baseProps);
+ LOGGER.trace(() -> "New or modified binding props: %s".formatted(newOrModifiedBindingProps));
+
+ Map mergedProps = new HashMap<>(newOrModifiedBinderProps);
+ mergedProps.putAll(newOrModifiedBindingProps);
+ LOGGER.trace(() -> "Final merged props: %s".formatted(mergedProps));
+
+ return mergedProps;
+ }
+
+ private static Map extractNewOrModifiedProperties(Map candidateProps,
+ Map baseProps) {
+ Map newOrModifiedProps = new HashMap<>();
+ candidateProps.forEach((propName, propValue) -> {
+ if (!baseProps.containsKey(propName) || (!Objects.equals(propValue, baseProps.get(propName)))) {
+ newOrModifiedProps.put(propName, propValue);
+ }
+ });
+ return newOrModifiedProps;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java
new file mode 100644
index 000000000..16de5c11c
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java
@@ -0,0 +1,317 @@
+/*
+ * Copyright 2022-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.Optional;
+import java.util.Set;
+
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.schema.SchemaType;
+
+import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
+import org.springframework.cloud.stream.binder.HeaderMode;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarProducerProperties;
+import org.springframework.cloud.stream.binder.pulsar.provisioning.PulsarTopicProvisioner;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
+import org.springframework.integration.support.management.ManageableLifecycle;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+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;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.core.SchemaResolver;
+import org.springframework.pulsar.core.TypedMessageBuilderCustomizer;
+import org.springframework.pulsar.listener.AbstractPulsarMessageListenerContainer;
+import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer;
+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.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+public class PulsarMessageChannelBinder extends
+ AbstractMessageChannelBinder, ExtendedProducerProperties, PulsarTopicProvisioner>
+ implements ExtendedPropertiesBinder {
+
+ private final PulsarTemplate pulsarTemplate;
+
+ private final PulsarConsumerFactory pulsarConsumerFactory;
+
+ private final PulsarBinderConfigurationProperties binderConfigProps;
+
+ private final SchemaResolver schemaResolver;
+
+ private final PulsarHeaderMapper headerMapper;
+
+ private PulsarExtendedBindingProperties extendedBindingProperties = new PulsarExtendedBindingProperties();
+
+ public PulsarMessageChannelBinder(PulsarTopicProvisioner provisioningProvider,
+ PulsarTemplate pulsarTemplate, PulsarConsumerFactory pulsarConsumerFactory,
+ PulsarBinderConfigurationProperties binderConfigProps, SchemaResolver schemaResolver,
+ PulsarHeaderMapper headerMapper) {
+ super(null, provisioningProvider);
+ this.pulsarTemplate = pulsarTemplate;
+ this.pulsarConsumerFactory = pulsarConsumerFactory;
+ this.binderConfigProps = binderConfigProps;
+ this.schemaResolver = schemaResolver;
+ this.headerMapper = headerMapper;
+ }
+
+ @Override
+ protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
+ ExtendedProducerProperties producerProperties, MessageChannel errorChannel) {
+ final Schema schema;
+ if (producerProperties.isUseNativeEncoding()) {
+ var schemaType = Optional.ofNullable(producerProperties.getExtension().getSchemaType())
+ .orElse(SchemaType.NONE);
+ schema = this.schemaResolver
+ .resolveSchema(schemaType, producerProperties.getExtension().getMessageType(),
+ producerProperties.getExtension().getMessageKeyType())
+ .orElseThrow(() -> "Could not determine producer schema for " + destination.getName());
+ }
+ else {
+ schema = null;
+ }
+ var baseProducerProps = new ProducerConfigProperties().buildProperties();
+ var binderProducerProps = this.binderConfigProps.getProducer().buildProperties();
+ var bindingProducerProps = producerProperties.getExtension().buildProperties();
+ var mergedProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseProducerProps,
+ binderProducerProps, bindingProducerProps);
+
+ var handler = new PulsarProducerConfigurationMessageHandler(this.pulsarTemplate, schema, destination.getName(),
+ (builder) -> ProducerBuilderConfigurationUtil.loadConf(builder, mergedProducerProps),
+ determineOutboundHeaderMapper(producerProperties));
+ handler.setApplicationContext(getApplicationContext());
+ handler.setBeanFactory(getBeanFactory());
+
+ return handler;
+ }
+
+ @Nullable
+ private PulsarBinderHeaderMapper determineOutboundHeaderMapper(
+ ExtendedProducerProperties extProducerProps) {
+ if (HeaderMode.none.equals(extProducerProps.getHeaderMode())) {
+ return null;
+ }
+ return new PulsarBinderHeaderMapper(this.headerMapper);
+ }
+
+ @Override
+ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group,
+ ExtendedConsumerProperties properties) {
+ var containerProperties = new PulsarContainerProperties();
+ containerProperties.setTopics(Set.of(destination.getName()));
+
+ var inboundHeaderMapper = determineInboundHeaderMapper(properties);
+
+ var messageDrivenChannelAdapter = new PulsarMessageDrivenChannelAdapter();
+ containerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, pulsarMsg) -> {
+ var springMessage = (inboundHeaderMapper != null)
+ ? MessageBuilder.createMessage(pulsarMsg.getValue(), inboundHeaderMapper.toSpringHeaders(pulsarMsg))
+ : MessageBuilder.withPayload(pulsarMsg.getValue()).build();
+ messageDrivenChannelAdapter.send(springMessage);
+ });
+
+ if (properties.isUseNativeDecoding()) {
+ var schemaType = Optional.ofNullable(properties.getExtension().getSchemaType()).orElse(SchemaType.NONE);
+ var schema = this.schemaResolver
+ .resolveSchema(schemaType, properties.getExtension().getMessageType(),
+ properties.getExtension().getMessageKeyType())
+ .orElseThrow(() -> "Could not determine consumer schema for " + destination.getName());
+ containerProperties.setSchema(schema);
+ }
+ else {
+ containerProperties.setSchema(Schema.BYTES);
+ }
+ var subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination);
+ containerProperties.setSubscriptionName(subscriptionName);
+
+ var baseConsumerProps = new ConsumerConfigProperties().buildProperties();
+ var binderConsumerProps = this.binderConfigProps.getConsumer().buildProperties();
+ var bindingConsumerProps = properties.getExtension().buildProperties();
+ var mergedConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseConsumerProps,
+ binderConsumerProps, bindingConsumerProps);
+ containerProperties.getPulsarConsumerProperties().putAll(mergedConsumerProps);
+ containerProperties.updateContainerProperties();
+
+ var container = new DefaultPulsarMessageListenerContainer<>(this.pulsarConsumerFactory, containerProperties);
+ messageDrivenChannelAdapter.setMessageListenerContainer(container);
+
+ return messageDrivenChannelAdapter;
+ }
+
+ @Nullable
+ private PulsarBinderHeaderMapper determineInboundHeaderMapper(
+ ExtendedConsumerProperties extConsumerProps) {
+ if (HeaderMode.none.equals(extConsumerProps.getHeaderMode())) {
+ return null;
+ }
+ return new PulsarBinderHeaderMapper(this.headerMapper);
+ }
+
+ @Override
+ public PulsarConsumerProperties getExtendedConsumerProperties(String channelName) {
+ return this.extendedBindingProperties.getExtendedConsumerProperties(channelName);
+ }
+
+ @Override
+ public PulsarProducerProperties getExtendedProducerProperties(String channelName) {
+ return this.extendedBindingProperties.getExtendedProducerProperties(channelName);
+ }
+
+ @Override
+ public String getDefaultsPrefix() {
+ return null;
+ }
+
+ @Override
+ public Class getExtendedPropertiesEntryClass() {
+ return null;
+ }
+
+ public PulsarExtendedBindingProperties getExtendedBindingProperties() {
+ return this.extendedBindingProperties;
+ }
+
+ public void setExtendedBindingProperties(PulsarExtendedBindingProperties extendedBindingProperties) {
+ this.extendedBindingProperties = extendedBindingProperties;
+ }
+
+ static class PulsarMessageDrivenChannelAdapter extends MessageProducerSupport {
+
+ AbstractPulsarMessageListenerContainer messageListenerContainer;
+
+ public void send(Message message) {
+ sendMessage(message);
+ }
+
+ @Override
+ protected void doStart() {
+ this.messageListenerContainer.start();
+ }
+
+ @Override
+ protected void doStop() {
+ this.messageListenerContainer.stop();
+ }
+
+ public void setMessageListenerContainer(AbstractPulsarMessageListenerContainer messageListenerContainer) {
+ this.messageListenerContainer = messageListenerContainer;
+ }
+
+ }
+
+ static class PulsarProducerConfigurationMessageHandler extends AbstractMessageProducingHandler
+ implements ManageableLifecycle {
+
+ private final PulsarTemplate pulsarTemplate;
+
+ private final Schema schema;
+
+ private final String destination;
+
+ private final ProducerBuilderCustomizer layeredProducerPropsCustomizer;
+
+ private final PulsarHeaderMapper headerMapper;
+
+ private boolean running = true;
+
+ PulsarProducerConfigurationMessageHandler(PulsarTemplate pulsarTemplate, Schema schema,
+ String destination, ProducerBuilderCustomizer layeredProducerPropsCustomizer,
+ PulsarHeaderMapper headerMapper) {
+ this.pulsarTemplate = pulsarTemplate;
+ this.schema = schema;
+ this.destination = destination;
+ this.layeredProducerPropsCustomizer = layeredProducerPropsCustomizer;
+ this.headerMapper = headerMapper;
+ }
+
+ @Override
+ public void start() {
+ try {
+ super.onInit();
+ }
+ catch (Exception ex) {
+ this.logger.error(ex, "Initialization errors: ");
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Override
+ public void stop() {
+ // TODO - should we close the underlyiung producer?
+ this.running = false;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return this.running;
+ }
+
+ @Override
+ protected void handleMessageInternal(Message message) {
+ try {
+ // @formatter:off
+ this.pulsarTemplate.newMessage(message.getPayload())
+ .withTopic(this.destination)
+ .withSchema(this.schema)
+ .withProducerCustomizer(this.layeredProducerPropsCustomizer)
+ .withMessageCustomizer(this.applySpringHeadersAsPulsarProperties(message.getHeaders()))
+ .sendAsync();
+ // @formatter:on
+ }
+ catch (PulsarClientException ex) {
+ logger.trace(ex, "Failed to send message to destination: " + this.destination);
+ }
+ }
+
+ private TypedMessageBuilderCustomizer applySpringHeadersAsPulsarProperties(MessageHeaders headers) {
+ return (mb) -> {
+ if (this.headerMapper != null) {
+ this.headerMapper.toPulsarHeaders(headers).forEach(mb::property);
+ }
+ };
+ }
+
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/config/PulsarBinderConfiguration.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/config/PulsarBinderConfiguration.java
new file mode 100644
index 000000000..6e77e06a6
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/config/PulsarBinderConfiguration.java
@@ -0,0 +1,75 @@
+/*
+ * 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.config;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.pulsar.PulsarMessageChannelBinder;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarExtendedBindingProperties;
+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;
+import org.springframework.pulsar.core.SchemaResolver;
+import org.springframework.pulsar.support.header.JacksonUtils;
+import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper;
+import org.springframework.pulsar.support.header.PulsarHeaderMapper;
+import org.springframework.pulsar.support.header.ToStringPulsarHeaderMapper;
+
+/**
+ * Pulsar binder {@link Configuration}.
+ *
+ * @author Soby Chacko
+ */
+@Configuration(proxyBeanMethods = false)
+@ConditionalOnMissingBean(Binder.class)
+@EnableConfigurationProperties({ PulsarProperties.class, PulsarExtendedBindingProperties.class,
+ PulsarBinderConfigurationProperties.class })
+public class PulsarBinderConfiguration {
+
+ @Bean
+ public PulsarTopicProvisioner pulsarTopicProvisioner(PulsarAdministration pulsarAdministration,
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties) {
+ return new PulsarTopicProvisioner(pulsarAdministration, pulsarBinderConfigurationProperties);
+ }
+
+ @Bean
+ @ConditionalOnMissingBean
+ public PulsarHeaderMapper pulsarHeaderMapper() {
+ if (JacksonUtils.isJacksonPresent()) {
+ return JsonPulsarHeaderMapper.builder().build();
+ }
+ return new ToStringPulsarHeaderMapper();
+ }
+
+ @Bean
+ public PulsarMessageChannelBinder pulsarMessageChannelBinder(PulsarTopicProvisioner pulsarTopicProvisioner,
+ PulsarTemplate pulsarTemplate, PulsarConsumerFactory pulsarConsumerFactory,
+ PulsarBinderConfigurationProperties binderConfigProps, PulsarExtendedBindingProperties bindingConfigProps,
+ SchemaResolver schemaResolver, PulsarHeaderMapper headerMapper) {
+ PulsarMessageChannelBinder pulsarMessageChannelBinder = new PulsarMessageChannelBinder(pulsarTopicProvisioner,
+ pulsarTemplate, pulsarConsumerFactory, binderConfigProps, schemaResolver, headerMapper);
+ pulsarMessageChannelBinder.setExtendedBindingProperties(bindingConfigProps);
+ return pulsarMessageChannelBinder;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/package-info.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/package-info.java
new file mode 100644
index 000000000..95966720b
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/package-info.java
@@ -0,0 +1,25 @@
+/*
+ * 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 containing Spring Cloud Stream binder classes for Apache Pulsar.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.cloud.stream.binder.pulsar;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarBinderConfigurationProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarBinderConfigurationProperties.java
new file mode 100644
index 000000000..7722b7c6e
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarBinderConfigurationProperties.java
@@ -0,0 +1,71 @@
+/*
+ * 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 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.
+ *
+ * These properties are applied at the binder level (to all bindings).
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@ConfigurationProperties(prefix = "spring.cloud.stream.pulsar.binder")
+public class PulsarBinderConfigurationProperties {
+
+ /**
+ * Pulsar consumer specific binder-level properties (applied to all bindings).
+ */
+ @NestedConfigurationProperty
+ private final ConsumerConfigProperties consumer = new ConsumerConfigProperties();
+
+ /**
+ * Pulsar producer specific binder-level properties (applied to all bindings).
+ */
+ @NestedConfigurationProperty
+ private final ProducerConfigProperties producer = new ProducerConfigProperties();
+
+ /**
+ * Number of topic partitions.
+ */
+ @Nullable
+ private Integer partitionCount;
+
+ public ConsumerConfigProperties getConsumer() {
+ return this.consumer;
+ }
+
+ public ProducerConfigProperties getProducer() {
+ return this.producer;
+ }
+
+ @Nullable
+ public Integer getPartitionCount() {
+ return this.partitionCount;
+ }
+
+ public void setPartitionCount(Integer partitionCount) {
+ this.partitionCount = partitionCount;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarBindingProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarBindingProperties.java
new file mode 100644
index 000000000..8abe593ca
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarBindingProperties.java
@@ -0,0 +1,71 @@
+/*
+ * 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 org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
+
+/**
+ * Container for Pulsar specific extended producer and consumer binding properties.
+ *
+ * These properties are applied to individual bindings and will override any binder-level
+ * setting.
+ *
+ *
+ * NOTE: This class is only referenced as a value in the
+ * {@link PulsarExtendedBindingProperties#getBindings() bindings map} and therefore, by
+ * default is not included in the generated configuration metadata. To get around this
+ * limitation it is annotated with {@code @ConfigurationProperties}. However, that is the
+ * only reason it is annotated and is not intended to be used directly.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@SuppressWarnings("ConfigurationProperties")
+@ConfigurationProperties("spring.cloud.stream.pulsar.bindings.for-docs-only")
+public class PulsarBindingProperties implements BinderSpecificPropertiesProvider {
+
+ /**
+ * Pulsar consumer specific binding properties.
+ */
+ @NestedConfigurationProperty
+ private PulsarConsumerProperties consumer = new PulsarConsumerProperties();
+
+ /**
+ * Pulsar producer specific binding properties.
+ */
+ @NestedConfigurationProperty
+ private PulsarProducerProperties producer = new PulsarProducerProperties();
+
+ public PulsarConsumerProperties getConsumer() {
+ return this.consumer;
+ }
+
+ public void setConsumer(PulsarConsumerProperties consumer) {
+ this.consumer = consumer;
+ }
+
+ public PulsarProducerProperties getProducer() {
+ return this.producer;
+ }
+
+ public void setProducer(PulsarProducerProperties producer) {
+ this.producer = producer;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarConsumerProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarConsumerProperties.java
new file mode 100644
index 000000000..ea4684aac
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarConsumerProperties.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2022-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 org.apache.pulsar.common.schema.SchemaType;
+
+import org.springframework.lang.Nullable;
+import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
+
+/**
+ * Pulsar consumer properties used by the binder.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+public class PulsarConsumerProperties extends ConsumerConfigProperties {
+
+ /**
+ * Pulsar {@link SchemaType} for this binding.
+ */
+ @Nullable
+ private SchemaType schemaType;
+
+ /**
+ * Pulsar message type for this binding.
+ */
+ @Nullable
+ private Class messageType;
+
+ /**
+ * Pulsar message key type for this binding (only used when schema type is
+ * {@code }KEY_VALUE}).
+ */
+ @Nullable
+ private Class messageKeyType;
+
+ /**
+ * Number of topic partitions.
+ */
+ @Nullable
+ private Integer partitionCount;
+
+ @Nullable
+ public SchemaType getSchemaType() {
+ return this.schemaType;
+ }
+
+ public void setSchemaType(@Nullable SchemaType schemaType) {
+ this.schemaType = schemaType;
+ }
+
+ @Nullable
+ public Class getMessageType() {
+ return this.messageType;
+ }
+
+ public void setMessageType(@Nullable Class messageType) {
+ this.messageType = messageType;
+ }
+
+ @Nullable
+ public Class getMessageKeyType() {
+ return this.messageKeyType;
+ }
+
+ public void setMessageKeyType(@Nullable Class messageKeyType) {
+ this.messageKeyType = messageKeyType;
+ }
+
+ @Nullable
+ public Integer getPartitionCount() {
+ return this.partitionCount;
+ }
+
+ public void setPartitionCount(Integer partitionCount) {
+ this.partitionCount = partitionCount;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarExtendedBindingProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarExtendedBindingProperties.java
new file mode 100644
index 000000000..4e487a3c0
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarExtendedBindingProperties.java
@@ -0,0 +1,60 @@
+/*
+ * 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.util.Map;
+
+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.
+ *
+ * These properties are applied to individual bindings and will override any binder-level
+ * settings.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@ConfigurationProperties("spring.cloud.stream.pulsar")
+public class PulsarExtendedBindingProperties extends
+ AbstractExtendedBindingProperties {
+
+ private static final String DEFAULTS_PREFIX = "spring.cloud.stream.pulsar.default";
+
+ @Override
+ public String getDefaultsPrefix() {
+ return DEFAULTS_PREFIX;
+ }
+
+ /**
+ * Properties per individual binding name (e.g. 'mySink-in-0').
+ */
+ @Override
+ public Map getBindings() {
+ return this.doGetBindings();
+ }
+
+ @Override
+ public Class getExtendedPropertiesEntryClass() {
+ return PulsarBindingProperties.class;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarProducerProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarProducerProperties.java
new file mode 100644
index 000000000..db73378e2
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarProducerProperties.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2022-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 org.apache.pulsar.common.schema.SchemaType;
+
+import org.springframework.lang.Nullable;
+import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
+
+/**
+ * Pulsar producer properties used by the binder.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+public class PulsarProducerProperties extends ProducerConfigProperties {
+
+ /**
+ * Pulsar {@link SchemaType} for this binding.
+ */
+ @Nullable
+ private SchemaType schemaType;
+
+ /**
+ * Pulsar message type for this binding.
+ */
+ @Nullable
+ private Class messageType;
+
+ /**
+ * Pulsar message key type for this binding (only used when schema type is
+ * {@code }KEY_VALUE}).
+ */
+ @Nullable
+ private Class messageKeyType;
+
+ /**
+ * Number of topic partitions.
+ */
+ @Nullable
+ private Integer partitionCount;
+
+ @Nullable
+ public SchemaType getSchemaType() {
+ return this.schemaType;
+ }
+
+ public void setSchemaType(@Nullable SchemaType schemaType) {
+ this.schemaType = schemaType;
+ }
+
+ @Nullable
+ public Class getMessageType() {
+ return this.messageType;
+ }
+
+ public void setMessageType(@Nullable Class messageType) {
+ this.messageType = messageType;
+ }
+
+ @Nullable
+ public Class getMessageKeyType() {
+ return this.messageKeyType;
+ }
+
+ public void setMessageKeyType(@Nullable Class messageKeyType) {
+ this.messageKeyType = messageKeyType;
+ }
+
+ @Nullable
+ public Integer getPartitionCount() {
+ return this.partitionCount;
+ }
+
+ public void setPartitionCount(Integer partitionCount) {
+ this.partitionCount = partitionCount;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/package-info.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/package-info.java
new file mode 100644
index 000000000..688bc2473
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/package-info.java
@@ -0,0 +1,25 @@
+/*
+ * 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 containing Spring Cloud Stream binder properties classes for Apache Pulsar.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.cloud.stream.binder.pulsar.properties;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/provisioning/PulsarTopicProvisioner.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/provisioning/PulsarTopicProvisioner.java
new file mode 100644
index 000000000..46a2f674e
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/provisioning/PulsarTopicProvisioner.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2022-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.provisioning;
+
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarProducerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.cloud.stream.provisioning.ProvisioningException;
+import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+import org.springframework.lang.Nullable;
+import org.springframework.pulsar.core.PulsarAdministration;
+import org.springframework.pulsar.core.PulsarTopic;
+
+/**
+ * Pulsar topic provisioner.
+ *
+ * @author Soby Chacko
+ */
+public class PulsarTopicProvisioner implements
+ ProvisioningProvider, ExtendedProducerProperties> {
+
+ private final PulsarAdministration pulsarAdministration;
+
+ private final PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties;
+
+ public PulsarTopicProvisioner(PulsarAdministration pulsarAdministration,
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties) {
+ this.pulsarAdministration = pulsarAdministration;
+ this.pulsarBinderConfigurationProperties = pulsarBinderConfigurationProperties;
+ }
+
+ @Override
+ public ProducerDestination provisionProducerDestination(String name,
+ ExtendedProducerProperties pulsarProducerProperties)
+ throws ProvisioningException {
+ Integer partitionCountFromBinding = pulsarProducerProperties.getExtension().getPartitionCount();
+ var partitionCount = getPartitionCount(partitionCountFromBinding);
+ var pulsarTopic = PulsarTopic.builder(name).numberOfPartitions(partitionCount).build();
+ this.pulsarAdministration.createOrModifyTopics(pulsarTopic);
+ return new PulsarDestination(pulsarTopic.topicName(), pulsarTopic.numberOfPartitions());
+ }
+
+ private int getPartitionCount(@Nullable Integer partitionCountConfig) {
+ var partitionCount = this.pulsarBinderConfigurationProperties.getPartitionCount();
+ if (partitionCountConfig != null && partitionCountConfig > 0) {
+ partitionCount = partitionCountConfig;
+ }
+ return partitionCount == null ? 0 : partitionCount;
+ }
+
+ @Override
+ public ConsumerDestination provisionConsumerDestination(String name, String group,
+ ExtendedConsumerProperties pulsarConsumerProperties)
+ throws ProvisioningException {
+ var partitionCountFromBinding = pulsarConsumerProperties.getExtension().getPartitionCount();
+ var partitionCount = getPartitionCount(partitionCountFromBinding);
+ var pulsarTopic = PulsarTopic.builder(name).numberOfPartitions(partitionCount).build();
+ this.pulsarAdministration.createOrModifyTopics(pulsarTopic);
+ return new PulsarDestination(pulsarTopic.topicName(), pulsarTopic.numberOfPartitions());
+ }
+
+ private record PulsarDestination(String destinationName,
+ Integer partitions) implements ProducerDestination, ConsumerDestination {
+
+ @Override
+ public String getName() {
+ return this.destinationName;
+ }
+
+ @Override
+ public String getNameForPartition(int partition) {
+ return this.destinationName;
+ }
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/provisioning/package-info.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/provisioning/package-info.java
new file mode 100644
index 000000000..c6a016779
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/provisioning/package-info.java
@@ -0,0 +1,25 @@
+/*
+ * 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 containing Spring Cloud Stream binder provisioning classes for Apache Pulsar.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.cloud.stream.binder.pulsar.provisioning;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/resources/META-INF/spring.binders b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/resources/META-INF/spring.binders
new file mode 100644
index 000000000..778e113a6
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/resources/META-INF/spring.binders
@@ -0,0 +1,2 @@
+pulsar:\
+org.springframework.cloud.stream.binder.pulsar.config.PulsarBinderConfiguration
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/AbstractPulsarTestBinder.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/AbstractPulsarTestBinder.java
new file mode 100644
index 000000000..a6191c2f7
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/AbstractPulsarTestBinder.java
@@ -0,0 +1,48 @@
+/*
+ * 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 org.springframework.cloud.stream.binder.AbstractPollableConsumerTestBinder;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarProducerProperties;
+import org.springframework.context.ApplicationContext;
+
+/**
+ * Base class for {@link PulsarTestBinder}.
+ *
+ * @author Soby Chacko
+ */
+public abstract class AbstractPulsarTestBinder extends
+ AbstractPollableConsumerTestBinder, ExtendedProducerProperties> {
+
+ private ApplicationContext applicationContext;
+
+ @Override
+ public void cleanup() {
+ }
+
+ protected final void setApplicationContext(ApplicationContext context) {
+ this.applicationContext = context;
+ }
+
+ public ApplicationContext getApplicationContext() {
+ return this.applicationContext;
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderConfigurationPropertiesTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderConfigurationPropertiesTests.java
new file mode 100644
index 000000000..a1b8cdffa
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderConfigurationPropertiesTests.java
@@ -0,0 +1,121 @@
+/*
+ * 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.ProducerAccessMode;
+import org.apache.pulsar.client.api.SubscriptionMode;
+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.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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+
+/**
+ * Tests for {@link PulsarBinderConfigurationProperties}.
+ *
+ * @author Chris Bono
+ */
+public class PulsarBinderConfigurationPropertiesTests {
+
+ private final PulsarBinderConfigurationProperties properties = new PulsarBinderConfigurationProperties();
+
+ private void bind(Map 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);
+ }
+
+ @Test
+ void producerProperties() {
+ // Only spot check a few values (PulsarPropertiesTests does the heavy lifting)
+ Map 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 producerProps = properties.getProducer().buildProperties();
+
+ // 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
+ }
+
+ @Test
+ void consumerProperties() {
+ // Only spot check a few values (PulsarPropertiesTests does the heavy lifting)
+ Map 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 consumerProps = properties.getConsumer().buildProperties();
+
+ // 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
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderHeaderMapperTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderHeaderMapperTests.java
new file mode 100644
index 000000000..6a2b5f1cf
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderHeaderMapperTests.java
@@ -0,0 +1,115 @@
+/*
+ * 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.Message;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.springframework.cloud.stream.binder.BinderHeaders;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.pulsar.support.header.PulsarHeaderMapper;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.entry;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link PulsarBinderHeaderMapper}.
+ *
+ * @author Chris Bono
+ */
+@ExtendWith(MockitoExtension.class)
+class PulsarBinderHeaderMapperTests {
+
+ @Mock
+ private PulsarHeaderMapper delegateMapper;
+
+ @InjectMocks
+ private PulsarBinderHeaderMapper binderHeaderMapper;
+
+ @Nested
+ class ToPulsarHeadersOutboundTests {
+
+ @Test
+ void delegateReturnsEmptyHeaders() {
+ var delegatePulsarHeaders = new HashMap();
+ when(delegateMapper.toPulsarHeaders(any(MessageHeaders.class))).thenReturn(delegatePulsarHeaders);
+ var springHeaders = mock(MessageHeaders.class);
+ var pulsarHeaders = binderHeaderMapper.toPulsarHeaders(springHeaders);
+ verify(delegateMapper).toPulsarHeaders(springHeaders);
+ assertThat(pulsarHeaders).isEmpty();
+ }
+
+ @Test
+ void neverHeadersRemovedFromDelegateHeaders() {
+ var delegatePulsarHeaders = new HashMap();
+ delegatePulsarHeaders.put(MessageHeaders.ID, "5150");
+ delegatePulsarHeaders.put(MessageHeaders.TIMESTAMP, "12345");
+ delegatePulsarHeaders.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, "5");
+ delegatePulsarHeaders.put(BinderHeaders.NATIVE_HEADERS_PRESENT, "true");
+ delegatePulsarHeaders.put("foo", "bar");
+ when(delegateMapper.toPulsarHeaders(any(MessageHeaders.class))).thenReturn(delegatePulsarHeaders);
+ var springHeaders = mock(MessageHeaders.class);
+ var pulsarHeaders = binderHeaderMapper.toPulsarHeaders(springHeaders);
+ verify(delegateMapper).toPulsarHeaders(springHeaders);
+ assertThat(pulsarHeaders).containsOnly(entry("foo", "bar"));
+ }
+
+ }
+
+ @Nested
+ class ToSpringHeadersInboundTests {
+
+ @Test
+ void delegateReturnsEmptyHeaders() {
+ var emptyDelegateHeaders = mock(MessageHeaders.class);
+ when(emptyDelegateHeaders.isEmpty()).thenReturn(true);
+ when(delegateMapper.toSpringHeaders(any(Message.class))).thenReturn(emptyDelegateHeaders);
+ var springHeaders = binderHeaderMapper.toSpringHeaders(mock(Message.class));
+ assertThat(springHeaders).isSameAs(emptyDelegateHeaders);
+ verify(springHeaders).isEmpty();
+ verifyNoMoreInteractions(springHeaders);
+ }
+
+ @Test
+ void nativeHeadersIndicatorAddedToDelegateHeaders() {
+ var delegateSpringHeaders = new MessageHeaders(Map.of("foo", "bar"));
+ when(delegateMapper.toSpringHeaders(any(Message.class))).thenReturn(delegateSpringHeaders);
+ var pulsarMessage = mock(Message.class);
+ var springHeaders = binderHeaderMapper.toSpringHeaders(pulsarMessage);
+ verify(delegateMapper).toSpringHeaders(pulsarMessage);
+ assertThat(springHeaders).containsEntry("foo", "bar").containsEntry(BinderHeaders.NATIVE_HEADERS_PRESENT,
+ Boolean.TRUE);
+
+ }
+
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java
new file mode 100644
index 000000000..c0030cf41
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java
@@ -0,0 +1,884 @@
+/*
+ * 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.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+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;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.system.CapturedOutput;
+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;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
+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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for {@link PulsarBinderIntegrationTests}.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@ExtendWith(OutputCaptureExtension.class)
+@SuppressWarnings("JUnitMalformedDeclaration")
+class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
+
+ private static final int AWAIT_DURATION = 10;
+
+ @Test
+ void binderAndBindingPropsAreAppliedAndRespected(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(BinderAndBindingPropsTestConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext context = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
+ "--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.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")) {
+
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: test-basic-scenario"));
+
+ // now verify the properties were set onto producer and consumer as expected
+ TrackingProducerFactory producerFactory = context.getBean(TrackingProducerFactory.class);
+ assertThat(producerFactory.producersCreated).isNotEmpty().element(0)
+ .hasFieldOrPropertyWithValue("producerName", "textSupplierProducer-fromBinding")
+ .hasFieldOrPropertyWithValue("conf.maxPendingMessages", 1100)
+ .hasFieldOrPropertyWithValue("conf.blockIfQueueFull", true);
+
+ TrackingConsumerFactory consumerFactory = context.getBean(TrackingConsumerFactory.class);
+ assertThat(consumerFactory.consumersCreated).isNotEmpty().element(0)
+ .hasFieldOrPropertyWithValue("consumerName", "textLogger-fromBinding")
+ .hasFieldOrPropertyWithValue("conf.subscriptionName", "textLoggerSub-fromBinder");
+ }
+ }
+
+ @Nested
+ class DefaultEncoding {
+
+ @Test
+ void primitiveTypeString(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(PrimitiveTextConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: test-basic-scenario"));
+ }
+ }
+
+ @Test
+ void primitiveTypeFloat(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(PrimitiveFloatConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: 3.14"));
+ }
+ }
+
+ }
+
+ @Nested
+ class NativeEncoding {
+
+ @Test
+ void primitiveTypeFloat(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(PrimitiveFloatConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: 3.14"));
+ }
+ }
+
+ @Test
+ void jsonTypeFooWithSchemaType(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(JsonFooConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.schema-type=JSON",
+ "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type="
+ + Foo.class.getName(),
+ "--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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: Foo[value=5150]"));
+ }
+ }
+
+ @Test
+ void jsonTypeFooWithoutSchemaTypeDefaultsToJsonSchema(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(JsonFooConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type="
+ + 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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: Foo[value=5150]"));
+ }
+ }
+
+ @Test
+ void avroTypeUserWithSchemaType(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(AvroUserConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.schema-type=AVRO",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type="
+ + User.class.getName(),
+ "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true",
+ "--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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}"));
+ }
+ }
+
+ @Test
+ void avroTypeUserWithoutSchemaTypeWithCustomMappingsViaProps(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(AvroUserConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type="
+ + User.class.getName(),
+ "--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.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))
+ .until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}"));
+ }
+ }
+
+ @Test
+ void avroTypeUserWithoutSchemaTypeWithCustomMappingsViaCustomizer(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(AvroUserConfigCustomMappings.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type="
+ + User.class.getName(),
+ "--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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}"));
+ }
+ }
+
+ @Test
+ void keyValueAvroTypeWithSchemaTypeAndCustomTypeMappingsViaProps(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(KeyValueAvroUserConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.schema-type=KEY_VALUE",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type="
+ + User.class.getName(),
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-key-type="
+ + String.class.getName(),
+ "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.schema-type=KEY_VALUE",
+ "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type="
+ + 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.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))
+ .until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}"));
+ }
+ }
+
+ @Test
+ void keyValueAvroTypeWithoutSchemaTypeAndCustomTypeMappingsViaProps(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(KeyValueAvroUserConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type="
+ + User.class.getName(),
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-key-type="
+ + String.class.getName(),
+ "--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.message-key-type="
+ + String.class.getName(),
+ "--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))
+ .until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}"));
+ }
+ }
+
+ @Test
+ void keyValueAvroTypeWithSchemaTypeAndCustomTypeMappingsViaCustomizer(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(KeyValueAvroUserConfigCustomMappings.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.schema-type=KEY_VALUE",
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type="
+ + User.class.getName(),
+ "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-key-type="
+ + String.class.getName(),
+ "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true",
+ "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.schema-type=KEY_VALUE",
+ "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type="
+ + 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")) {
+ Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION))
+ .until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}"));
+ }
+ }
+
+ @Test
+ void keyValueJsonTypeWithoutSchemaTypeAndWithoutCustomTypeMappings(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(KeyValueJsonFooConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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",
+ "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true",
+ "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type="
+ + Foo.class.getName(),
+ "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-key-type="
+ + String.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.message-key-type="
+ + String.class.getName(),
+ "--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]"));
+ }
+ }
+
+ }
+
+ @Nested
+ class CustomMessageHeaders {
+
+ @Test
+ void headersPropagatedSendAndReceive(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(CustomSimpleHeadersConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ // 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"));
+ }
+ }
+
+ @Test
+ void complexHeadersAreEncodedAndPropagated(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(CustomComplexHeadersConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ // 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]"));
+ }
+ }
+
+ @Test
+ void producerHeaderModeNone(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(CustomComplexHeadersConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ // 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"));
+ }
+ }
+
+ @Test
+ void consumerHeaderModeNone(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(CustomComplexHeadersConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ // 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"));
+ }
+ }
+
+ @Test
+ void customHeaderMapperRespected(CapturedOutput output) {
+ SpringApplication app = new SpringApplication(CustomHeaderMapperConfig.class);
+ app.setWebApplicationType(WebApplicationType.NONE);
+ try (ConfigurableApplicationContext ignored = app.run(
+ "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+ "--spring.pulsar.administration.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")) {
+ // 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]"));
+ }
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class CustomSimpleHeadersConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ private int msgCount = 0;
+
+ @Bean
+ public Supplier> springMessageSupplier() {
+ return () -> {
+ msgCount++;
+ return MessageBuilder.withPayload("test-headers-msg-" + msgCount)
+ .setHeader("custom-id", "5150-" + msgCount).build();
+ };
+ }
+
+ @Bean
+ public Consumer> springMessageLogger() {
+ return s -> this.logger.info("Hello binder: {} w/ custom-id: {}", s.getPayload(),
+ s.getHeaders().get("custom-id"));
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class CustomComplexHeadersConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ private int msgCount = 0;
+
+ @Bean
+ public Supplier> springMessageSupplier() {
+ return () -> {
+ msgCount++;
+ return MessageBuilder.withPayload("test-headers-msg-" + msgCount)
+ .setHeader("custom-id", new FooHeader("5150-" + msgCount)).build();
+ };
+ }
+
+ @Bean
+ public Consumer> springMessageLogger() {
+ return s -> {
+ var header = s.getHeaders().get("custom-id");
+ if (header != null) {
+ assertThat(header).isInstanceOf(FooHeader.class);
+ }
+ this.logger.info("Hello binder: {} w/ custom-id: {}", s.getPayload(), header);
+ };
+ }
+
+ record FooHeader(String value) {
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class CustomHeaderMapperConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ private int msgCount = 0;
+
+ @Bean
+ public PulsarHeaderMapper extendedToStringHeaderMapper() {
+ return new ToStringPulsarHeaderMapper(List.of("custom-id"), List.of("foo", "custom-id")) {
+ @Override
+ public Map toPulsarHeaders(MessageHeaders springHeaders) {
+ Map pulsarHeaders = super.toPulsarHeaders(springHeaders);
+ // foo and custom-id are allowed and expected
+ assertThat(pulsarHeaders).containsKeys("foo", "custom-id");
+ return pulsarHeaders;
+ }
+
+ @Override
+ public MessageHeaders toSpringHeaders(org.apache.pulsar.client.api.Message pulsarMessage) {
+ MessageHeaders springHeaders = super.toSpringHeaders(pulsarMessage);
+ // foo not allowed, custom-id allowed
+ assertThat(springHeaders).doesNotContainKey("foo").containsKey("custom-id");
+ return springHeaders;
+ }
+
+ @Override
+ protected String toPulsarHeaderValue(String name, Object value, Object context) {
+ return "tph->" + super.toPulsarHeaderValue(name, value, context);
+ }
+
+ @Override
+ protected Object toSpringHeaderValue(String headerName, String rawHeader, Object context) {
+ return "tsh->" + super.toSpringHeaderValue(headerName, rawHeader, context);
+ }
+ };
+ }
+
+ @Bean
+ public Supplier> springMessageSupplier() {
+ return () -> {
+ msgCount++;
+ return MessageBuilder.withPayload("test-headers-msg-" + msgCount)
+ .setHeader("foo", "bar-" + msgCount)
+ .setHeader("custom-id", new FooHeader("5150-" + msgCount)).build();
+ };
+ }
+
+ @Bean
+ public Consumer> springMessageLogger() {
+ return s -> {
+ var header = s.getHeaders().get("custom-id");
+ if (header != null) {
+ assertThat(header).isInstanceOf(String.class);
+ }
+ var fooHeader = s.getHeaders().get("foo");
+ assertThat(fooHeader).isNull();
+ this.logger.info("Hello binder: {} w/ custom-id: {}", s.getPayload(), header);
+ };
+ }
+
+ record FooHeader(String value) {
+ }
+
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class PrimitiveTextConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Bean
+ public Supplier textSupplier() {
+ return () -> "test-basic-scenario";
+ }
+
+ @Bean
+ public Consumer textLogger() {
+ return s -> this.logger.info("Hello binder: " + s);
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ @Import(PrimitiveTextConfig.class)
+ static class BinderAndBindingPropsTestConfig {
+
+ @Bean
+ public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient,
+ PulsarProperties pulsarProperties, TopicResolver topicResolver) {
+ return new TrackingProducerFactory(pulsarClient, pulsarProperties.buildProducerProperties(), topicResolver);
+ }
+
+ @Bean
+ public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient,
+ PulsarProperties pulsarProperties) {
+ return new TrackingConsumerFactory(pulsarClient, pulsarProperties.buildConsumerProperties());
+ }
+
+ }
+
+ static class TrackingProducerFactory extends DefaultPulsarProducerFactory {
+
+ List> producersCreated = new ArrayList<>();
+
+ TrackingProducerFactory(PulsarClient pulsarClient, Map config, TopicResolver topicResolver) {
+ super(pulsarClient, config, topicResolver);
+ }
+
+ @Override
+ protected Producer doCreateProducer(Schema schema, @Nullable String topic,
+ @Nullable Collection encryptionKeys,
+ @Nullable List> producerBuilderCustomizers)
+ throws PulsarClientException {
+ Producer producer = super.doCreateProducer(schema, topic, encryptionKeys,
+ producerBuilderCustomizers);
+ producersCreated.add(producer);
+ return producer;
+ }
+
+ }
+
+ static class TrackingConsumerFactory extends DefaultPulsarConsumerFactory {
+
+ List> consumersCreated = new ArrayList<>();
+
+ TrackingConsumerFactory(PulsarClient pulsarClient, Map consumerConfig) {
+ super(pulsarClient, consumerConfig);
+ }
+
+ @Override
+ public org.apache.pulsar.client.api.Consumer createConsumer(Schema schema,
+ @Nullable Collection topics, @Nullable String subscriptionName,
+ @Nullable Map metadataProperties,
+ @Nullable List> consumerBuilderCustomizers)
+ throws PulsarClientException {
+ org.apache.pulsar.client.api.Consumer consumer = super.createConsumer(schema, topics,
+ subscriptionName, metadataProperties, consumerBuilderCustomizers);
+ consumersCreated.add(consumer);
+ return consumer;
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class PrimitiveFloatConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Bean
+ public Supplier piSupplier() {
+ return () -> 3.14f;
+ }
+
+ @Bean
+ public Consumer piLogger() {
+ return f -> this.logger.info("Hello binder: " + f);
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class JsonFooConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Bean
+ public Supplier fooSupplier() {
+ return () -> new Foo("5150");
+ }
+
+ @Bean
+ public Consumer fooLogger() {
+ return f -> this.logger.info("Hello binder: " + f);
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ @Import(JsonFooConfig.class)
+ static class JsonFooWithCustomMappingConfig {
+
+ @Bean
+ public SchemaResolverCustomizer customMappings() {
+ return (resolver) -> resolver.addCustomSchemaMapping(Foo.class, JSONSchema.of(Foo.class));
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class AvroUserConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Bean
+ public Supplier userSupplier() {
+ return () -> new User("user21", 21);
+ }
+
+ @Bean
+ public Consumer userLogger() {
+ return f -> this.logger.info("Hello binder: " + f);
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ @Import(AvroUserConfig.class)
+ static class AvroUserConfigCustomMappings {
+
+ @Bean
+ public SchemaResolverCustomizer customMappings() {
+ return (resolver) -> resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class KeyValueAvroUserConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Bean
+ public Supplier> userSupplier() {
+ return () -> new KeyValue<>("21", new User("user21", 21));
+ }
+
+ @Bean
+ public Consumer> userLogger() {
+ return f -> this.logger.info("Hello binder: " + f.getKey() + "->" + f.getValue());
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ @Import(KeyValueAvroUserConfig.class)
+ static class KeyValueAvroUserConfigCustomMappings {
+
+ @Bean
+ public SchemaResolverCustomizer customMappings() {
+ return (resolver) -> resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
+ }
+
+ }
+
+ @EnableAutoConfiguration
+ @SpringBootConfiguration
+ static class KeyValueJsonFooConfig {
+
+ private final Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Bean
+ public Supplier> fooSupplier() {
+ return () -> new KeyValue<>("5150", new Foo("5150"));
+ }
+
+ @Bean
+ public Consumer> fooLogger() {
+ return f -> this.logger.info("Hello binder: " + f.getKey() + "->" + f.getValue());
+ }
+
+ }
+
+ record Foo(String value) {
+ }
+
+ /**
+ * Do not convert this to a Record as Avro does not seem to work well w/ records.
+ */
+ static class User {
+
+ private String name;
+
+ private int age;
+
+ User() {
+ }
+
+ User(String name, int age) {
+ this.name = name;
+ this.age = age;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public int getAge() {
+ return age;
+ }
+
+ public void setAge(int age) {
+ this.age = age;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ User user = (User) o;
+ return age == user.age && Objects.equals(name, user.name);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, age);
+ }
+
+ @Override
+ public String toString() {
+ return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
+ }
+
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java
new file mode 100644
index 000000000..100b75138
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java
@@ -0,0 +1,255 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
+import org.springframework.cloud.stream.binder.Spy;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarProducerProperties;
+import org.springframework.cloud.stream.binder.pulsar.provisioning.PulsarTopicProvisioner;
+import org.springframework.cloud.stream.config.BindingProperties;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
+import org.springframework.pulsar.core.PulsarAdministration;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeTypeUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link PulsarMessageChannelBinder}.
+ *
+ * @author Soby Chacko
+ */
+public class PulsarBinderTests extends
+ PartitionCapableBinderTests, ExtendedProducerProperties>
+ implements PulsarTestContainerSupport {
+
+ private PulsarTestBinder binder;
+
+ @Nullable
+ protected PulsarClient pulsarClient;
+
+ @BeforeEach
+ void createPulsarClient() throws PulsarClientException {
+ pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+ }
+
+ @AfterEach
+ void closePulsarClient() throws PulsarClientException {
+ if (pulsarClient != null && !pulsarClient.isClosed()) {
+ pulsarClient.close();
+ }
+ }
+
+ @Override
+ protected boolean usesExplicitRouting() {
+ return false;
+ }
+
+ @Override
+ protected String getClassUnderTestName() {
+ return PulsarMessageChannelBinder.class.getSimpleName();
+ }
+
+ @Override
+ protected PulsarTestBinder getBinder() {
+ var pulsarAdministration = new PulsarAdministration(
+ Map.of("serviceUrl", PulsarTestContainerSupport.getHttpServiceUrl()));
+ var configProps = new PulsarBinderConfigurationProperties();
+ var provisioner = new PulsarTopicProvisioner(pulsarAdministration, configProps);
+ var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap());
+ var pulsarTemplate = new PulsarTemplate<>(producerFactory);
+ var config = Map.of("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest);
+ var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, config);
+ if (this.binder == null) {
+ this.binder = new PulsarTestBinder(provisioner, pulsarTemplate, consumerFactory, configProps,
+ new DefaultSchemaResolver(), JsonPulsarHeaderMapper.builder().build());
+ }
+ return this.binder;
+ }
+
+ @Override
+ protected ExtendedConsumerProperties createConsumerProperties() {
+ final ExtendedConsumerProperties pulsarConsumerProperties = new ExtendedConsumerProperties<>(
+ new PulsarConsumerProperties());
+ return pulsarConsumerProperties;
+ }
+
+ @Override
+ public Spy spyOn(String name) {
+ return null;
+ }
+
+ private ExtendedProducerProperties createProducerProperties() {
+ return this.createProducerProperties(null);
+ }
+
+ @Override
+ protected ExtendedProducerProperties createProducerProperties(TestInfo testInto) {
+ return new ExtendedProducerProperties<>(new PulsarProducerProperties());
+ }
+
+ @Override
+ protected void binderBindUnbindLatency() throws InterruptedException {
+ Thread.sleep(500);
+ }
+
+ @Test
+ @Override
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ public void testSendAndReceive(TestInfo testInfo) throws Exception {
+ Binder binder = getBinder();
+ BindingProperties outputBindingProperties = createProducerBindingProperties(createProducerProperties());
+
+ DirectChannel moduleOutputChannel = createBindableChannel("output", outputBindingProperties);
+ ExtendedConsumerProperties consumerProperties = createConsumerProperties();
+ DirectChannel moduleInputChannel = createBindableChannel("input",
+ createConsumerBindingProperties(consumerProperties));
+
+ Binding producerBinding = binder.bindProducer("foo.bar", moduleOutputChannel,
+ outputBindingProperties.getProducer());
+ Binding consumerBinding = binder.bindConsumer("foo.bar", null, moduleInputChannel,
+ consumerProperties);
+
+ Message message = MessageBuilder
+ .withPayload("foo".getBytes(StandardCharsets.UTF_8)).build();
+
+ // Let the consumer actually bind to the producer before sending a msg
+ binderBindUnbindLatency();
+ moduleOutputChannel.send(message);
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference> inboundMessageRef = new AtomicReference<>();
+ moduleInputChannel.subscribe(message1 -> {
+ try {
+ inboundMessageRef.set((Message) message1);
+ }
+ finally {
+ latch.countDown();
+ }
+ });
+ Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message");
+
+ assertThat(inboundMessageRef.get()).isNotNull();
+ assertThat(new String(inboundMessageRef.get().getPayload(), StandardCharsets.UTF_8)).isEqualTo("foo");
+
+ producerBinding.unbind();
+ consumerBinding.unbind();
+ }
+
+ @Test
+ @Override
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ public void testAnonymousGroup(TestInfo testInfo) throws Exception {
+ Binder binder = getBinder();
+ BindingProperties producerBindingProperties = createProducerBindingProperties(
+ createProducerProperties(testInfo));
+ DirectChannel output = createBindableChannel("output", producerBindingProperties);
+ Binding producerBinding = binder.bindProducer(
+ String.format("defaultGroup%s0", getDestinationNameDelimiter()), output,
+ producerBindingProperties.getProducer());
+
+ QueueChannel input1 = new QueueChannel();
+ Binding binding1 = binder.bindConsumer(
+ String.format("defaultGroup%s0", getDestinationNameDelimiter()), null, input1,
+ createConsumerProperties());
+
+ QueueChannel input2 = new QueueChannel();
+ Binding binding2 = binder.bindConsumer(
+ String.format("defaultGroup%s0", getDestinationNameDelimiter()), null, input2,
+ createConsumerProperties());
+
+ String testPayload1 = "foo-" + UUID.randomUUID();
+ output.send(MessageBuilder.withPayload(testPayload1)
+ .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build());
+
+ Message receivedMessage1 = (Message) receive(input1);
+ assertThat(receivedMessage1).isNotNull();
+ assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1);
+
+ Message receivedMessage2 = (Message) receive(input2);
+ assertThat(receivedMessage2).isNotNull();
+ assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1);
+
+ binding2.unbind();
+
+ String testPayload2 = "foo-" + UUID.randomUUID();
+ output.send(MessageBuilder.withPayload(testPayload2)
+ .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build());
+
+ binding2 = binder.bindConsumer(String.format("defaultGroup%s0", getDestinationNameDelimiter()), null, input2,
+ createConsumerProperties());
+ String testPayload3 = "foo-" + UUID.randomUUID();
+ output.send(MessageBuilder.withPayload(testPayload3)
+ .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build());
+
+ receivedMessage1 = (Message) receive(input1);
+ assertThat(receivedMessage1).isNotNull();
+ assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2);
+ receivedMessage1 = (Message) receive(input1);
+ assertThat(receivedMessage1).isNotNull();
+ assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload3);
+
+ receivedMessage2 = (Message) receive(input2);
+ assertThat(receivedMessage2).isNotNull();
+ assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1);
+
+ producerBinding.unbind();
+ binding1.unbind();
+ binding2.unbind();
+ }
+
+ @Test
+ @Override
+ @Disabled
+ public void testPartitionedModuleSpEL(TestInfo testInfo) {
+ // This use-case needs to be further evaluated for Pulsar binder.
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java
new file mode 100644
index 000000000..c23a425d8
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java
@@ -0,0 +1,133 @@
+/*
+ * 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.Collections;
+import java.util.Map;
+import java.util.stream.Stream;
+
+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.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link PulsarBinderUtils}.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+public class PulsarBinderUtilsTests {
+
+ @Nested
+ class SubscriptionNameTests {
+
+ @Test
+ void respectsValueWhenSetAsProperty() {
+ var consumerDestination = mock(ConsumerDestination.class);
+ var pulsarConsumerProperties = mock(PulsarConsumerProperties.class);
+ when(pulsarConsumerProperties.getSubscriptionName()).thenReturn("my-sub");
+ assertThat(PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination))
+ .isEqualTo("my-sub");
+ }
+
+ @Test
+ void generatesValueWhenNotSetAsProperty() {
+ var consumerDestination = mock(ConsumerDestination.class);
+ var pulsarConsumerProperties = mock(PulsarConsumerProperties.class);
+ when(pulsarConsumerProperties.getSubscriptionName()).thenReturn(null);
+ when(consumerDestination.getName()).thenReturn("my-topic");
+ assertThat(PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination))
+ .startsWith("my-topic-anon-subscription-");
+ }
+
+ }
+
+ @Nested
+ class MergedPropertiesTests {
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("mergePropertiesTestProvider")
+ void mergePropertiesTest(String testName, Map baseProps, Map binderProps,
+ Map bindingProps, Map expectedMergedProps) {
+ assertThat(PulsarBinderUtils.mergePropertiesWithPrecedence(baseProps, binderProps, bindingProps))
+ .containsExactlyInAnyOrderEntriesOf(expectedMergedProps);
+ }
+
+ // @formatter:off
+ static Stream mergePropertiesTestProvider() {
+ return Stream.of(
+ arguments("binderLevelContainsSamePropAsBaseWithDiffValue",
+ Map.of("foo", "foo-base"),
+ Map.of("foo", "foo-binder"),
+ Collections.emptyMap(),
+ Map.of("foo", "foo-binder")),
+ arguments("binderLevelContainsNewPropNotInBase",
+ Collections.emptyMap(),
+ Map.of("foo", "foo-binder"),
+ Collections.emptyMap(),
+ Map.of("foo", "foo-binder")),
+ arguments("binderLevelContainsSamePropAsBaseWithSameValue",
+ Map.of("foo", "foo-base"),
+ Map.of("foo", "foo-base"),
+ Collections.emptyMap(),
+ Collections.emptyMap()),
+ arguments("bindingLevelContainsSamePropAsBaseWithDiffValue",
+ Map.of("foo", "foo-base"),
+ Collections.emptyMap(),
+ Map.of("foo", "foo-binding"),
+ Map.of("foo", "foo-binding")),
+ arguments("bindingLevelContainsNewPropNotInBase",
+ Collections.emptyMap(),
+ Map.of("foo", "foo-binding"),
+ Collections.emptyMap(),
+ Map.of("foo", "foo-binding")),
+ arguments("bindingLevelContainsSamePropAsBaseWithSameValue",
+ Map.of("foo", "foo-base"),
+ Collections.emptyMap(),
+ Map.of("foo", "foo-base"),
+ Collections.emptyMap()),
+ arguments("bindingOverridesBinder",
+ Map.of("bar", "bar-base"),
+ Map.of("foo", "foo-binder"),
+ Map.of("foo", "foo-binding"),
+ Map.of("foo", "foo-binding")),
+ arguments("binderOverridesBaseAndBindingOverridesBinder",
+ Map.of("foo", "foo-base"),
+ Map.of("foo", "foo-binder"),
+ Map.of("foo", "foo-binding"),
+ Map.of("foo", "foo-binding")),
+ arguments("onlyBaseProps",
+ Map.of("foo", "foo-base"),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap()));
+ }
+ // @formatter:on
+
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingPropertiesTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingPropertiesTests.java
new file mode 100644
index 000000000..38a68a729
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingPropertiesTests.java
@@ -0,0 +1,137 @@
+/*
+ * 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.ProducerAccessMode;
+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.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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+
+/**
+ * Tests for {@link PulsarExtendedBindingProperties}.
+ *
+ * @author Chris Bono
+ */
+public class PulsarExtendedBindingPropertiesTests {
+
+ private final PulsarExtendedBindingProperties properties = new PulsarExtendedBindingProperties();
+
+ private void bind(Map 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 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");
+
+ bind(props);
+
+ assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
+ Map producerProps = properties.getExtendedProducerProperties("my-foo").buildProperties();
+ // 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
+ }
+
+ @Test
+ void consumerProperties() {
+ // Only spot check a few values (PulsarPropertiesTests does the heavy lifting)
+ Map 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");
+
+ bind(props);
+
+ assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
+ Map consumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties();
+ // 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
+ }
+
+ @Test
+ void extendedBindingsArePropagatedToContainerProperties() {
+ Map 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");
+
+ bind(props);
+
+ var bindingConsumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties();
+ PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
+ pulsarContainerProperties.getPulsarConsumerProperties().putAll(bindingConsumerProps);
+
+ 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);
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTestBinder.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTestBinder.java
new file mode 100644
index 000000000..6040c54cc
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTestBinder.java
@@ -0,0 +1,63 @@
+/*
+ * 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 org.springframework.cloud.stream.binder.pulsar.properties.PulsarBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.pulsar.provisioning.PulsarTopicProvisioner;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.core.SchemaResolver;
+import org.springframework.pulsar.support.header.PulsarHeaderMapper;
+
+/**
+ * Test binder to exercise producer/consumer bindings in
+ * {@link PulsarMessageChannelBinder}.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+public class PulsarTestBinder extends AbstractPulsarTestBinder {
+
+ @SuppressWarnings({ "unchecked" })
+ PulsarTestBinder(PulsarTopicProvisioner pulsarTopicProvisioner, PulsarTemplate pulsarTemplate,
+ PulsarConsumerFactory pulsarConsumerFactory, PulsarBinderConfigurationProperties binderConfigProps,
+ SchemaResolver schemaResolver, PulsarHeaderMapper headerMapper) {
+
+ try {
+ var binder = new PulsarMessageChannelBinder(pulsarTopicProvisioner, (PulsarTemplate) pulsarTemplate,
+ pulsarConsumerFactory, binderConfigProps, schemaResolver, headerMapper);
+ var context = new AnnotationConfigApplicationContext(Config.class);
+ setApplicationContext(context);
+ binder.setApplicationContext(context);
+ binder.afterPropertiesSet();
+ this.setPollableConsumerBinder(binder);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Configuration
+ @EnableIntegration
+ static class Config {
+
+ }
+
+}
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTestContainerSupport.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTestContainerSupport.java
new file mode 100644
index 000000000..b8bddb838
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTestContainerSupport.java
@@ -0,0 +1,68 @@
+/*
+ * 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.Locale;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * Provides a static {@link PulsarContainer} that can be shared across test classes.
+ *
+ * @author Chris Bono
+ */
+@Testcontainers(disabledWithoutDocker = true)
+public interface PulsarTestContainerSupport {
+
+ PulsarContainer PULSAR_CONTAINER = new PulsarContainer(getPulsarImage());
+
+ @BeforeAll
+ static void startContainer() {
+ PULSAR_CONTAINER.start();
+ }
+
+ static String getPulsarBrokerUrl() {
+ return PULSAR_CONTAINER.getPulsarBrokerUrl();
+ }
+
+ static DockerImageName getPulsarImage() {
+ return isRunningOnMacM1() ? getMacM1PulsarImage() : getStandardPulsarImage();
+ }
+
+ 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");
+ }
+
+}
+
diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTopicProvisionerTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTopicProvisionerTests.java
new file mode 100644
index 000000000..fb60f0e71
--- /dev/null
+++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTopicProvisionerTests.java
@@ -0,0 +1,136 @@
+/*
+ * 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 org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarProducerProperties;
+import org.springframework.cloud.stream.binder.pulsar.provisioning.PulsarTopicProvisioner;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.pulsar.core.PulsarAdministration;
+import org.springframework.pulsar.core.PulsarTopic;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author Soby Chacko
+ */
+public class PulsarTopicProvisionerTests {
+
+ @Test
+ void provisionThroughProducerBindingWithDefaultPartitioning() {
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties = new PulsarBinderConfigurationProperties();
+ PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
+ PulsarTopicProvisioner pulsarTopicProvisioner = new PulsarTopicProvisioner(pulsarAdministration,
+ pulsarBinderConfigurationProperties);
+ ExtendedProducerProperties properties = new ExtendedProducerProperties<>(
+ new PulsarProducerProperties());
+ ProducerDestination producerDestination = pulsarTopicProvisioner.provisionProducerDestination("foo",
+ properties);
+ verifyAndAssert(pulsarAdministration, producerDestination.getName(), "foo", 0);
+ }
+
+ private static void verifyAndAssert(PulsarAdministration pulsarAdministration, String actualProducerDestination,
+ String expectedProducerDestination, int expectedPartitionCount) {
+ ArgumentCaptor pulsarTopicArgumentCaptor = ArgumentCaptor.forClass(PulsarTopic.class);
+ verify(pulsarAdministration, times(1)).createOrModifyTopics(pulsarTopicArgumentCaptor.capture());
+ assertThat(actualProducerDestination).isEqualTo(expectedProducerDestination);
+ PulsarTopic pulsarTopic = pulsarTopicArgumentCaptor.getValue();
+ assertThat(pulsarTopic.topicName()).isEqualTo(expectedProducerDestination);
+ assertThat(pulsarTopic.numberOfPartitions()).isEqualTo(expectedPartitionCount);
+ }
+
+ @Test
+ void provisionThroughConsumerBindingWithDefaultPartitioning() {
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties = new PulsarBinderConfigurationProperties();
+ PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
+ PulsarTopicProvisioner pulsarTopicProvisioner = new PulsarTopicProvisioner(pulsarAdministration,
+ pulsarBinderConfigurationProperties);
+ ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(
+ new PulsarConsumerProperties());
+ ConsumerDestination consumerDestination = pulsarTopicProvisioner.provisionConsumerDestination("bar", "",
+ properties);
+ verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "bar", 0);
+ }
+
+ @Test
+ void provisioningOnProducerBindingWithPartitionsSetAtTheBinderProperties() {
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties = new PulsarBinderConfigurationProperties();
+ pulsarBinderConfigurationProperties.setPartitionCount(4);
+ PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
+ PulsarTopicProvisioner pulsarTopicProvisioner = new PulsarTopicProvisioner(pulsarAdministration,
+ pulsarBinderConfigurationProperties);
+ ExtendedProducerProperties properties = new ExtendedProducerProperties<>(
+ new PulsarProducerProperties());
+ ProducerDestination producerDestination = pulsarTopicProvisioner.provisionProducerDestination("foo",
+ properties);
+ verifyAndAssert(pulsarAdministration, producerDestination.getName(), "foo", 4);
+ }
+
+ @Test
+ void provisioningOnProducerBindingWithPartitionsSetAtTheBindingProperties() {
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties = new PulsarBinderConfigurationProperties();
+ PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
+ PulsarTopicProvisioner pulsarTopicProvisioner = new PulsarTopicProvisioner(pulsarAdministration,
+ pulsarBinderConfigurationProperties);
+ ExtendedProducerProperties properties = new ExtendedProducerProperties<>(
+ new PulsarProducerProperties());
+ properties.getExtension().setPartitionCount(4);
+ ProducerDestination producerDestination = pulsarTopicProvisioner.provisionProducerDestination("foo",
+ properties);
+ verifyAndAssert(pulsarAdministration, producerDestination.getName(), "foo", 4);
+ }
+
+ @Test
+ void provisionThroughConsumerBindingWithPartitionsSetAtTheBinderProperties() {
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties = new PulsarBinderConfigurationProperties();
+ pulsarBinderConfigurationProperties.setPartitionCount(4);
+ PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
+ PulsarTopicProvisioner pulsarTopicProvisioner = new PulsarTopicProvisioner(pulsarAdministration,
+ pulsarBinderConfigurationProperties);
+ ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(
+ new PulsarConsumerProperties());
+ ConsumerDestination consumerDestination = pulsarTopicProvisioner.provisionConsumerDestination("bar", "",
+ properties);
+ verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "bar", 4);
+ }
+
+ @Test
+ void provisionThroughConsumerBindingWithPartitionsSetAtTheBindingProperties() {
+ PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties = new PulsarBinderConfigurationProperties();
+ PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
+ PulsarTopicProvisioner pulsarTopicProvisioner = new PulsarTopicProvisioner(pulsarAdministration,
+ pulsarBinderConfigurationProperties);
+ PulsarConsumerProperties pulsarConsumerProperties = new PulsarConsumerProperties();
+ pulsarConsumerProperties.setPartitionCount(4);
+ ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(
+ pulsarConsumerProperties);
+ ConsumerDestination consumerDestination = pulsarTopicProvisioner.provisionConsumerDestination("bar", "",
+ properties);
+ verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "bar", 4);
+ }
+
+}
diff --git a/bom/pom.xml b/bom/pom.xml
index 312710f74..ca9b97131 100644
--- a/bom/pom.xml
+++ b/bom/pom.xml
@@ -7,7 +7,7 @@
pom
spring-cloud-stream-release-build
Spring Cloud Stream Release Build
- 4.0.3-SNAPSHOT
+ 4.1.0-SNAPSHOT
org.springframework.cloud
spring-cloud-build
diff --git a/bom/spring-cloud-starter-parent/pom.xml b/bom/spring-cloud-starter-parent/pom.xml
index 6f2bf2da6..fd511e54a 100644
--- a/bom/spring-cloud-starter-parent/pom.xml
+++ b/bom/spring-cloud-starter-parent/pom.xml
@@ -11,7 +11,7 @@
org.springframework.cloud
spring-cloud-stream-starter-parent
- 4.0.3-SNAPSHOT
+ 4.1.0-SNAPSHOT
spring-cloud-stream-starter-parent
Specifies Boot version for the releaser
pom
diff --git a/bom/spring-cloud-stream-dependencies/pom.xml b/bom/spring-cloud-stream-dependencies/pom.xml
index a79f767e7..065099be8 100644
--- a/bom/spring-cloud-stream-dependencies/pom.xml
+++ b/bom/spring-cloud-stream-dependencies/pom.xml
@@ -10,7 +10,7 @@
spring-cloud-stream-dependencies
- 4.0.3-SNAPSHOT
+ 4.1.0-SNAPSHOT
pom
spring-cloud-stream-dependencies
Spring Cloud Stream Dependencies