Subscription Type Propagation in the Binder

Author: sobychacko

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

@@ -20,6 +20,7 @@
 
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.schema.SchemaType;
 
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -162,6 +163,9 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
 		}
 		var subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination);
 		containerProperties.setSubscriptionName(subscriptionName);
+		if (properties.getExtension().getSubscriptionType() != SubscriptionType.Exclusive) {
+			containerProperties.setSubscriptionType(properties.getExtension().getSubscriptionType());
+		}
 
 		var baseConsumerProps = new ConsumerConfigProperties().buildProperties();
 		var binderConsumerProps = this.binderConfigProps.getConsumer().buildProperties();