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..229372d7c
--- /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,207 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+import org.springframework.boot.context.properties.PropertyMapper;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarConsumerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
+import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
+import org.springframework.util.StringUtils;
+import org.springframework.util.unit.DataSize;
+
+/**
+ * Binder utility methods.
+ *
+ * @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;
+ }
+
+ /**
+ * Gets a map representation of a {@link ProducerConfigProperties}.
+ * @param producerProps the producer props
+ * @return map representation of producer props where each entry is a field and its
+ * associated value
+ */
+ static Map convertProducerPropertiesToMap(ProducerConfigProperties producerProps) {
+ var properties = new PulsarBinderUtils.Properties();
+ var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+ map.from(producerProps::getTopicName).to(properties.in("topicName"));
+ map.from(producerProps::getProducerName).to(properties.in("producerName"));
+ map.from(producerProps::getSendTimeout).asInt(Duration::toMillis).to(properties.in("sendTimeoutMs"));
+ map.from(producerProps::getBlockIfQueueFull).to(properties.in("blockIfQueueFull"));
+ map.from(producerProps::getMaxPendingMessages).to(properties.in("maxPendingMessages"));
+ map.from(producerProps::getMaxPendingMessagesAcrossPartitions)
+ .to(properties.in("maxPendingMessagesAcrossPartitions"));
+ map.from(producerProps::getMessageRoutingMode).to(properties.in("messageRoutingMode"));
+ map.from(producerProps::getHashingScheme).to(properties.in("hashingScheme"));
+ map.from(producerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction"));
+ map.from(producerProps::getBatchingMaxPublishDelay).as(it -> it.toNanos() / 1000)
+ .to(properties.in("batchingMaxPublishDelayMicros"));
+ map.from(producerProps::getBatchingPartitionSwitchFrequencyByPublishDelay)
+ .to(properties.in("batchingPartitionSwitchFrequencyByPublishDelay"));
+ map.from(producerProps::getBatchingMaxMessages).to(properties.in("batchingMaxMessages"));
+ map.from(producerProps::getBatchingMaxBytes).asInt(DataSize::toBytes).to(properties.in("batchingMaxBytes"));
+ map.from(producerProps::getBatchingEnabled).to(properties.in("batchingEnabled"));
+ map.from(producerProps::getChunkingEnabled).to(properties.in("chunkingEnabled"));
+ map.from(producerProps::getEncryptionKeys).to(properties.in("encryptionKeys"));
+ map.from(producerProps::getCompressionType).to(properties.in("compressionType"));
+ map.from(producerProps::getInitialSequenceId).to(properties.in("initialSequenceId"));
+ map.from(producerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions"));
+ map.from(producerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
+ .to(properties.in("autoUpdatePartitionsIntervalSeconds"));
+ map.from(producerProps::getMultiSchema).to(properties.in("multiSchema"));
+ map.from(producerProps::getProducerAccessMode).to(properties.in("accessMode"));
+ map.from(producerProps::getLazyStartPartitionedProducers).to(properties.in("lazyStartPartitionedProducers"));
+ map.from(producerProps::getProperties).to(properties.in("properties"));
+ return properties;
+ }
+
+ /**
+ * Gets a map representation of a {@link ConsumerConfigProperties}.
+ * @param consumerProps the consumer props
+ * @return map representation of consumer props where each entry is a field and its
+ * associated value
+ */
+ static Map convertConsumerPropertiesToMap(ConsumerConfigProperties consumerProps) {
+ var properties = new PulsarBinderUtils.Properties();
+ var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+ map.from(consumerProps::getTopics).to(properties.in("topicNames"));
+ map.from(consumerProps::getTopicsPattern).to(properties.in("topicsPattern"));
+ map.from(consumerProps::getSubscriptionName).to(properties.in("subscriptionName"));
+ map.from(consumerProps::getSubscriptionType).to(properties.in("subscriptionType"));
+ map.from(consumerProps::getSubscriptionProperties).to(properties.in("subscriptionProperties"));
+ map.from(consumerProps::getSubscriptionMode).to(properties.in("subscriptionMode"));
+ map.from(consumerProps::getReceiverQueueSize).to(properties.in("receiverQueueSize"));
+ map.from(consumerProps::getAcknowledgementsGroupTime).as(it -> it.toNanos() / 1000)
+ .to(properties.in("acknowledgementsGroupTimeMicros"));
+ map.from(consumerProps::getNegativeAckRedeliveryDelay).as(it -> it.toNanos() / 1000)
+ .to(properties.in("negativeAckRedeliveryDelayMicros"));
+ map.from(consumerProps::getMaxTotalReceiverQueueSizeAcrossPartitions)
+ .to(properties.in("maxTotalReceiverQueueSizeAcrossPartitions"));
+ map.from(consumerProps::getConsumerName).to(properties.in("consumerName"));
+ map.from(consumerProps::getAckTimeout).as(Duration::toMillis).to(properties.in("ackTimeoutMillis"));
+ map.from(consumerProps::getTickDuration).as(Duration::toMillis).to(properties.in("tickDurationMillis"));
+ map.from(consumerProps::getPriorityLevel).to(properties.in("priorityLevel"));
+ map.from(consumerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction"));
+ map.from(consumerProps::getProperties).to(properties.in("properties"));
+ map.from(consumerProps::getReadCompacted).to(properties.in("readCompacted"));
+ map.from(consumerProps::getSubscriptionInitialPosition).to(properties.in("subscriptionInitialPosition"));
+ map.from(consumerProps::getPatternAutoDiscoveryPeriod).to(properties.in("patternAutoDiscoveryPeriod"));
+ map.from(consumerProps::getRegexSubscriptionMode).to(properties.in("regexSubscriptionMode"));
+ map.from(consumerProps::getDeadLetterPolicy).to(properties.in("deadLetterPolicy"));
+ map.from(consumerProps::getRetryEnable).to(properties.in("retryEnable"));
+ map.from(consumerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions"));
+ map.from(consumerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
+ .to(properties.in("autoUpdatePartitionsIntervalSeconds"));
+ map.from(consumerProps::getReplicateSubscriptionState).to(properties.in("replicateSubscriptionState"));
+ map.from(consumerProps::getResetIncludeHead).to(properties.in("resetIncludeHead"));
+ map.from(consumerProps::getBatchIndexAckEnabled).to(properties.in("batchIndexAckEnabled"));
+ map.from(consumerProps::getAckReceiptEnabled).to(properties.in("ackReceiptEnabled"));
+ map.from(consumerProps::getPoolMessages).to(properties.in("poolMessages"));
+ map.from(consumerProps::getStartPaused).to(properties.in("startPaused"));
+ map.from(consumerProps::getAutoAckOldestChunkedMessageOnQueueFull)
+ .to(properties.in("autoAckOldestChunkedMessageOnQueueFull"));
+ map.from(consumerProps::getMaxPendingChunkedMessage).to(properties.in("maxPendingChunkedMessage"));
+ map.from(consumerProps::getExpireTimeOfIncompleteChunkedMessage).as(Duration::toMillis)
+ .to(properties.in("expireTimeOfIncompleteChunkedMessageMillis"));
+ return properties;
+ }
+
+ static class Properties extends HashMap {
+
+ java.util.function.Consumer in(String key) {
+ return (value) -> put(key, value);
+ }
+
+ }
+
+}
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..d6eafe2eb
--- /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,320 @@
+/*
+ * 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