Port Pulsar binder auto-config changes from 0.2.x (#2724)

* Port "Use builder to autoconfigure PulsarProducerFactory"

- Syncs changes from spring-pulsar/0.2.x/8bee5ef6e907b80933419f639f9f1bb6a3669ec5

* Port "Use builder to autoconfigure PulsarConsumerFactory"

- Syncs changes from spring-pulsar/0.2.x @ 82458872b52024f5464e677bf21763ccf564c0d2

* Port "Use builder to autoconfigure PulsarAdministration"

- Syncs changes from spring-pulsar/0.2.x @ d38cc6cb8d1a1d12499e302cbee75c00211e3537

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

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

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

@@ -111,9 +111,11 @@ public class PulsarMessageChannelBinder extends
 		else {
 			schema = null;
 		}
-		var baseProducerProps = new ProducerConfigProperties().buildProperties();
-		var binderProducerProps = this.binderConfigProps.getProducer().buildProperties();
-		var bindingProducerProps = producerProperties.getExtension().buildProperties();
+
+		var baseProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(new ProducerConfigProperties());
+		var binderProducerProps = PulsarBinderUtils
+				.convertProducerPropertiesToMap(this.binderConfigProps.getProducer());
+		var bindingProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(producerProperties.getExtension());
 		var mergedProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseProducerProps,
 				binderProducerProps, bindingProducerProps);
 
@@ -165,9 +167,10 @@ public class PulsarMessageChannelBinder extends
 		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 baseConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(new ConsumerConfigProperties());
+		var binderConsumerProps = PulsarBinderUtils
+				.convertConsumerPropertiesToMap(this.binderConfigProps.getConsumer());
+		var bindingConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getExtension());
 		var mergedConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseConsumerProps,
 				binderConsumerProps, bindingConsumerProps);
 		containerProperties.getPulsarConsumerProperties().putAll(mergedConsumerProps);

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderConfigurationPropertiesTests.java

@@ -68,7 +68,7 @@ public class PulsarBinderConfigurationPropertiesTests {
 		props.put("spring.cloud.stream.pulsar.binder.producer.properties[my-prop]", "my-prop-value");
 
 		bind(props);
-		Map<String, Object> producerProps = properties.getProducer().buildProperties();
+		Map<String, Object> producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties.getProducer());
 
 		// Verify that the props can be loaded in a ProducerBuilder
 		assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
@@ -98,7 +98,7 @@ public class PulsarBinderConfigurationPropertiesTests {
 		props.put("spring.cloud.stream.pulsar.binder.consumer.receiver-queue-size", "1");
 
 		bind(props);
-		Map<String, Object> consumerProps = properties.getConsumer().buildProperties();
+		Map<String, Object> consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getConsumer());
 
 		// Verify that the props can be loaded in a ConsumerBuilder
 		assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps,

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

@@ -638,16 +638,23 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
 	@Import(PrimitiveTextConfig.class)
 	static class BinderAndBindingPropsTestConfig {
 
+		@SuppressWarnings("unchecked")
 		@Bean
 		public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient,
 				PulsarProperties pulsarProperties, TopicResolver topicResolver) {
-			return new TrackingProducerFactory(pulsarClient, pulsarProperties.buildProducerProperties(), topicResolver);
+			var customizer = (ProducerBuilderCustomizer<String>) pulsarProperties.getProducer()
+					.toProducerBuilderCustomizer();
+			return new TrackingProducerFactory(pulsarClient, pulsarProperties.getProducer().getTopicName(), customizer,
+					topicResolver);
 		}
 
+		@SuppressWarnings("unchecked")
 		@Bean
 		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
 				PulsarProperties pulsarProperties) {
-			return new TrackingConsumerFactory(pulsarClient, pulsarProperties.buildConsumerProperties());
+			var customizer = (ConsumerBuilderCustomizer<String>) pulsarProperties.getConsumer()
+					.toConsumerBuilderCustomizer();
+			return new TrackingConsumerFactory(pulsarClient, customizer);
 		}
 
 	}
@@ -656,8 +663,9 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
 
 		List<Producer<String>> producersCreated = new ArrayList<>();
 
-		TrackingProducerFactory(PulsarClient pulsarClient, Map<String, Object> config, TopicResolver topicResolver) {
-			super(pulsarClient, config, topicResolver);
+		TrackingProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
+				ProducerBuilderCustomizer<String> defaultConfigCustomizer, TopicResolver topicResolver) {
+			super(pulsarClient, defaultTopic, defaultConfigCustomizer, topicResolver);
 		}
 
 		@Override
@@ -677,8 +685,8 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
 
 		List<org.apache.pulsar.client.api.Consumer<String>> consumersCreated = new ArrayList<>();
 
-		TrackingConsumerFactory(PulsarClient pulsarClient, Map<String, Object> consumerConfig) {
-			super(pulsarClient, consumerConfig);
+		TrackingConsumerFactory(PulsarClient pulsarClient, ConsumerBuilderCustomizer<String> defaultConsumerConfig) {
+			super(pulsarClient, defaultConsumerConfig);
 		}
 
 		@Override

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java

@@ -17,8 +17,6 @@
 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;
@@ -100,14 +98,13 @@ public class PulsarBinderTests extends
 
 	@Override
 	protected PulsarTestBinder getBinder() {
-		var pulsarAdministration = new PulsarAdministration(
-				Map.of("serviceUrl", PulsarTestContainerSupport.getHttpServiceUrl()));
+		var pulsarAdministration = new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
 		var configProps = new PulsarBinderConfigurationProperties();
 		var provisioner = new PulsarTopicProvisioner(pulsarAdministration, configProps);
-		var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap());
+		var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 		var pulsarTemplate = new PulsarTemplate<>(producerFactory);
-		var config = Map.<String, Object>of("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest);
-		var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, config);
+		var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
+				(consumerBuilder -> consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)));
 		if (this.binder == null) {
 			this.binder = new PulsarTestBinder(provisioner, pulsarTemplate, consumerFactory, configProps,
 					new DefaultSchemaResolver(), JsonPulsarHeaderMapper.builder().build());

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

@@ -17,19 +17,42 @@
 package org.springframework.cloud.stream.binder.pulsar;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Stream;
 
+import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
+import org.apache.pulsar.client.api.HashingScheme;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.ProducerAccessMode;
+import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import org.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.PulsarConsumerProperties;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
+import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -130,4 +153,172 @@ public class PulsarBinderUtilsTests {
 
 	}
 
+	@Nested
+	class ConvertedPropertiesTests {
+
+		private final ProducerConfigProperties properties = new ProducerConfigProperties();
+
+		private void bind(Map<String, String> map) {
+			ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
+			new Binder(source).bind("spring.pulsar.producer", Bindable.ofInstance(this.properties));
+		}
+
+		@Test
+		void producerPropertiesToMap() {
+			Map<String, String> props = new HashMap<>();
+			props.put("spring.pulsar.producer.topic-name", "my-topic");
+			props.put("spring.pulsar.producer.producer-name", "my-producer");
+			props.put("spring.pulsar.producer.send-timeout", "2s");
+			props.put("spring.pulsar.producer.block-if-queue-full", "true");
+			props.put("spring.pulsar.producer.max-pending-messages", "3");
+			props.put("spring.pulsar.producer.max-pending-messages-across-partitions", "4");
+			props.put("spring.pulsar.producer.message-routing-mode", "custompartition");
+			props.put("spring.pulsar.producer.hashing-scheme", "murmur3_32hash");
+			props.put("spring.pulsar.producer.crypto-failure-action", "send");
+			props.put("spring.pulsar.producer.batching-max-publish-delay", "5s");
+			props.put("spring.pulsar.producer.batching-partition-switch-frequency-by-publish-delay", "6");
+			props.put("spring.pulsar.producer.batching-max-messages", "7");
+			props.put("spring.pulsar.producer.batching-max-bytes", "8");
+			props.put("spring.pulsar.producer.batching-enabled", "false");
+			props.put("spring.pulsar.producer.chunking-enabled", "true");
+			props.put("spring.pulsar.producer.encryption-keys[0]", "my-key");
+			props.put("spring.pulsar.producer.compression-type", "lz4");
+			props.put("spring.pulsar.producer.initial-sequence-id", "9");
+			props.put("spring.pulsar.producer.producer-access-mode", "exclusive");
+			props.put("spring.pulsar.producer.lazy-start=partitioned-producers", "true");
+			props.put("spring.pulsar.producer.properties[my-prop]", "my-prop-value");
+
+			bind(props);
+			Map<String, Object> producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties);
+
+			// Verify that the props can be loaded in a ProducerBuilder
+			assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
+					new ProducerConfigurationData(), ProducerConfigurationData.class));
+
+			assertThat(producerProps).containsEntry("topicName", "my-topic")
+					.containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000)
+					.containsEntry("blockIfQueueFull", true).containsEntry("maxPendingMessages", 3)
+					.containsEntry("maxPendingMessagesAcrossPartitions", 4)
+					.containsEntry("messageRoutingMode", MessageRoutingMode.CustomPartition)
+					.containsEntry("hashingScheme", HashingScheme.Murmur3_32Hash)
+					.containsEntry("cryptoFailureAction", ProducerCryptoFailureAction.SEND)
+					.containsEntry("batchingMaxPublishDelayMicros", 5_000_000L)
+					.containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6)
+					.containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8)
+					.containsEntry("batchingEnabled", false).containsEntry("chunkingEnabled", true)
+					.hasEntrySatisfying("encryptionKeys",
+							keys -> assertThat(keys).asInstanceOf(InstanceOfAssertFactories.collection(String.class))
+									.containsExactly("my-key"))
+					.containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L)
+					.containsEntry("accessMode", ProducerAccessMode.Exclusive)
+					.containsEntry("lazyStartPartitionedProducers", true).hasEntrySatisfying("properties",
+							properties -> assertThat(properties)
+									.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
+									.containsEntry("my-prop", "my-prop-value"));
+		}
+
+	}
+
+	@Nested
+	class ConvertedConsumerPropertiesTests {
+
+		private final ConsumerConfigProperties properties = new ConsumerConfigProperties();
+
+		private void bind(Map<String, String> map) {
+			ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
+			new Binder(source).bind("spring.pulsar.consumer", Bindable.ofInstance(this.properties));
+		}
+
+		@Test
+		void consumerPropertiesToMap() {
+			Map<String, String> props = new HashMap<>();
+			props.put("spring.pulsar.consumer.topics[0]", "my-topic");
+			props.put("spring.pulsar.consumer.topics-pattern", "my-pattern");
+			props.put("spring.pulsar.consumer.subscription-name", "my-subscription");
+			props.put("spring.pulsar.consumer.subscription-type", "shared");
+			props.put("spring.pulsar.consumer.subscription-properties[my-sub-prop]", "my-sub-prop-value");
+			props.put("spring.pulsar.consumer.subscription-mode", "nondurable");
+			props.put("spring.pulsar.consumer.receiver-queue-size", "1");
+			props.put("spring.pulsar.consumer.acknowledgements-group-time", "2s");
+			props.put("spring.pulsar.consumer.negative-ack-redelivery-delay", "3s");
+			props.put("spring.pulsar.consumer.max-total-receiver-queue-size-across-partitions", "5");
+			props.put("spring.pulsar.consumer.consumer-name", "my-consumer");
+			props.put("spring.pulsar.consumer.ack-timeout", "6s");
+			props.put("spring.pulsar.consumer.tick-duration", "7s");
+			props.put("spring.pulsar.consumer.priority-level", "8");
+			props.put("spring.pulsar.consumer.crypto-failure-action", "discard");
+			props.put("spring.pulsar.consumer.properties[my-prop]", "my-prop-value");
+			props.put("spring.pulsar.consumer.read-compacted", "true");
+			props.put("spring.pulsar.consumer.subscription-initial-position", "earliest");
+			props.put("spring.pulsar.consumer.pattern-auto-discovery-period", "9");
+			props.put("spring.pulsar.consumer.regex-subscription-mode", "all-topics");
+			props.put("spring.pulsar.consumer.dead-letter-policy.max-redeliver-count", "4");
+			props.put("spring.pulsar.consumer.dead-letter-policy.retry-letter-topic", "my-retry-topic");
+			props.put("spring.pulsar.consumer.dead-letter-policy.dead-letter-topic", "my-dlt-topic");
+			props.put("spring.pulsar.consumer.dead-letter-policy.initial-subscription-name", "my-initial-subscription");
+			props.put("spring.pulsar.consumer.retry-enable", "true");
+			props.put("spring.pulsar.consumer.auto-update-partitions", "false");
+			props.put("spring.pulsar.consumer.auto-update-partitions-interval", "10s");
+			props.put("spring.pulsar.consumer.replicate-subscription-state", "true");
+			props.put("spring.pulsar.consumer.reset-include-head", "true");
+			props.put("spring.pulsar.consumer.batch-index-ack-enabled", "true");
+			props.put("spring.pulsar.consumer.ack-receipt-enabled", "true");
+			props.put("spring.pulsar.consumer.pool-messages", "true");
+			props.put("spring.pulsar.consumer.start-paused", "true");
+			props.put("spring.pulsar.consumer.auto-ack-oldest-chunked-message-on-queue-full", "false");
+			props.put("spring.pulsar.consumer.max-pending-chunked-message", "11");
+			props.put("spring.pulsar.consumer.expire-time-of-incomplete-chunked-message", "12s");
+
+			bind(props);
+			Map<String, Object> consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties);
+
+			// Verify that the props can be loaded in a ConsumerBuilder
+			assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps,
+					new ConsumerConfigurationData<>(), ConsumerConfigurationData.class));
+
+			assertThat(consumerProps)
+					.hasEntrySatisfying("topicNames",
+							topics -> assertThat(topics)
+									.asInstanceOf(InstanceOfAssertFactories.collection(String.class))
+									.containsExactly("my-topic"))
+					.hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern"))
+					.containsEntry("subscriptionName", "my-subscription")
+					.containsEntry("subscriptionType", SubscriptionType.Shared)
+					.hasEntrySatisfying("subscriptionProperties",
+							properties -> assertThat(properties)
+									.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
+									.containsEntry("my-sub-prop", "my-sub-prop-value"))
+					.containsEntry("subscriptionMode", SubscriptionMode.NonDurable)
+					.containsEntry("receiverQueueSize", 1).containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L)
+					.containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L)
+					.containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5)
+					.containsEntry("consumerName", "my-consumer").containsEntry("ackTimeoutMillis", 6_000L)
+					.containsEntry("tickDurationMillis", 7_000L).containsEntry("priorityLevel", 8)
+					.containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD)
+					.hasEntrySatisfying("properties",
+							properties -> assertThat(properties)
+									.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
+									.containsEntry("my-prop", "my-prop-value"))
+					.containsEntry("readCompacted", true)
+					.containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest)
+					.containsEntry("patternAutoDiscoveryPeriod", 9)
+					.containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics)
+					.hasEntrySatisfying("deadLetterPolicy", dlp -> {
+						DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) dlp;
+						assertThat(deadLetterPolicy.getMaxRedeliverCount()).isEqualTo(4);
+						assertThat(deadLetterPolicy.getRetryLetterTopic()).isEqualTo("my-retry-topic");
+						assertThat(deadLetterPolicy.getDeadLetterTopic()).isEqualTo("my-dlt-topic");
+						assertThat(deadLetterPolicy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription");
+					}).containsEntry("retryEnable", true).containsEntry("autoUpdatePartitions", false)
+					.containsEntry("autoUpdatePartitionsIntervalSeconds", 10L)
+					.containsEntry("replicateSubscriptionState", true).containsEntry("resetIncludeHead", true)
+					.containsEntry("batchIndexAckEnabled", true).containsEntry("ackReceiptEnabled", true)
+					.containsEntry("poolMessages", true).containsEntry("startPaused", true)
+					.containsEntry("autoAckOldestChunkedMessageOnQueueFull", false)
+					.containsEntry("maxPendingChunkedMessage", 11)
+					.containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L);
+		}
+
+	}
+
 }

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingPropertiesTests.java

@@ -65,7 +65,8 @@ public class PulsarExtendedBindingPropertiesTests {
 		bind(props);
 
 		assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
-		Map<String, Object> producerProps = properties.getExtendedProducerProperties("my-foo").buildProperties();
+		Map<String, Object> producerProps = PulsarBinderUtils
+				.convertProducerPropertiesToMap(properties.getExtendedProducerProperties("my-foo"));
 		// Verify that the props can be loaded in a ProducerBuilder
 		assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
 				new ProducerConfigurationData(), ProducerConfigurationData.class));
@@ -95,7 +96,8 @@ public class PulsarExtendedBindingPropertiesTests {
 		bind(props);
 
 		assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
-		Map<String, Object> consumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties();
+		Map<String, Object> consumerProps = PulsarBinderUtils
+				.convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo"));
 		// Verify that the props can be loaded in a ConsumerBuilder
 		assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps,
 				new ConsumerConfigurationData<>(), ConsumerConfigurationData.class));
@@ -121,7 +123,8 @@ public class PulsarExtendedBindingPropertiesTests {
 
 		bind(props);
 
-		var bindingConsumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties();
+		var bindingConsumerProps = PulsarBinderUtils
+				.convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo"));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.getPulsarConsumerProperties().putAll(bindingConsumerProps);
 

buildPulsarBinder.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+
+if [ "$#" -eq 1 ]; then
+  ARGVALUE=$1
+  if [[ $ARGVALUE == *"skipTest"* ]]; then
+    echo "Skippping Tests"
+    ./mvnw clean install -f binders/pulsar-binder/pom.xml -DskipTests
+  elif [[ $ARGVALUE == *"disable.checks"* ]]; then
+    echo "Skippping checkstyle checks"
+    ./mvnw clean install -f binders/pulsar-binder/pom.xml -Ddisable.checks=true
+  fi
+else
+  ./mvnw clean install -f binders/pulsar-binder/pom.xml
+fi