Add Spring Pulsar transaction support

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarAutoConfiguration.java

@@ -28,12 +28,14 @@
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.thread.Threading;
 import org.springframework.boot.util.LambdaSafe;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.core.env.Environment;
@@ -57,6 +59,8 @@
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
 import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
+import org.springframework.pulsar.transaction.PulsarAwareTransactionManager;
+import org.springframework.pulsar.transaction.PulsarTransactionManager;
 
 /**
  * {@link EnableAutoConfiguration Auto-configuration} for Apache Pulsar.
@@ -126,8 +130,11 @@ private void applyProducerBuilderCustomizers(List<ProducerBuilderCustomizer<?>>
 	PulsarTemplate<?> pulsarTemplate(PulsarProducerFactory<?> pulsarProducerFactory,
 			ObjectProvider<ProducerInterceptor> producerInterceptors, SchemaResolver schemaResolver,
 			TopicResolver topicResolver) {
-		return new PulsarTemplate<>(pulsarProducerFactory, producerInterceptors.orderedStream().toList(),
-				schemaResolver, topicResolver, this.properties.getTemplate().isObservationsEnabled());
+		PulsarTemplate<?> template = new PulsarTemplate<>(pulsarProducerFactory,
+				producerInterceptors.orderedStream().toList(), schemaResolver, topicResolver,
+				this.properties.getTemplate().isObservationsEnabled());
+		this.propertiesMapper.customizeTemplate(template);
+		return template;
 	}
 
 	@Bean
@@ -142,6 +149,13 @@ DefaultPulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
 		return new DefaultPulsarConsumerFactory<>(pulsarClient, lambdaSafeCustomizers);
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	@Conditional(TransactionsEnabledCondition.class)
+	public PulsarAwareTransactionManager pulsarTransactionManager(PulsarClient pulsarClient) {
+		return new PulsarTransactionManager(pulsarClient);
+	}
+
 	@SuppressWarnings("unchecked")
 	private void applyConsumerBuilderCustomizers(List<ConsumerBuilderCustomizer<?>> customizers,
 			ConsumerBuilder<?> builder) {
@@ -153,13 +167,15 @@ private void applyConsumerBuilderCustomizers(List<ConsumerBuilderCustomizer<?>>
 	@ConditionalOnMissingBean(name = "pulsarListenerContainerFactory")
 	ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 			PulsarConsumerFactory<Object> pulsarConsumerFactory, SchemaResolver schemaResolver,
-			TopicResolver topicResolver, Environment environment) {
+			TopicResolver topicResolver, ObjectProvider<PulsarAwareTransactionManager> pulsarTransactionManager,
+			Environment environment) {
 		PulsarContainerProperties containerProperties = new PulsarContainerProperties();
 		containerProperties.setSchemaResolver(schemaResolver);
 		containerProperties.setTopicResolver(topicResolver);
 		if (Threading.VIRTUAL.isActive(environment)) {
 			containerProperties.setConsumerTaskExecutor(new VirtualThreadTaskExecutor("pulsar-consumer-"));
 		}
+		pulsarTransactionManager.ifUnique(containerProperties.transactions()::setTransactionManager);
 		this.propertiesMapper.customizeContainerProperties(containerProperties);
 		return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties);
 	}
@@ -203,4 +219,26 @@ static class EnablePulsarConfiguration {
 
 	}
 
+	/**
+	 * Custom condition that is true when transactions have been enabled in the template
+	 * and/or in the listener.
+	 */
+	static final class TransactionsEnabledCondition extends AnyNestedCondition {
+
+		TransactionsEnabledCondition() {
+			super(ConfigurationPhase.REGISTER_BEAN);
+		}
+
+		@ConditionalOnProperty(prefix = "spring.pulsar.template.transaction", name = "enabled", havingValue = "true")
+		static class TemplateTransactionEnabledCondition {
+
+		}
+
+		@ConditionalOnProperty(prefix = "spring.pulsar.listener.transaction", name = "enabled", havingValue = "true")
+		static class ListenerTransactionEnabledCondition {
+
+		}
+
+	}
+
 }

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarProperties.java

@@ -36,6 +36,7 @@
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import org.springframework.boot.context.properties.source.InvalidConfigurationPropertyValueException;
 import org.springframework.util.Assert;
 
 /**
@@ -103,6 +104,14 @@ public Template getTemplate() {
 		return this.template;
 	}
 
+	/**
+	 * Whether transactions are enabled for either the template or the listener.
+	 * @return whether transactions are enabled for either the template or the listener
+	 */
+	boolean isTransactionEnabled() {
+		return this.template.getTransaction().isEnabled() || this.listener.getTransaction().isEnabled();
+	}
+
 	public static class Client {
 
 		/**
@@ -763,6 +772,11 @@ public static class Listener {
 		 */
 		private boolean observationEnabled;
 
+		/**
+		 * Transaction settings.
+		 */
+		private final Transaction transaction = new ListenerTransaction();
+
 		public SchemaType getSchemaType() {
 			return this.schemaType;
 		}
@@ -779,6 +793,10 @@ public void setObservationEnabled(boolean observationEnabled) {
 			this.observationEnabled = observationEnabled;
 		}
 
+		public Transaction getTransaction() {
+			return this.transaction;
+		}
+
 	}
 
 	public static class Reader {
@@ -858,6 +876,11 @@ public static class Template {
 		 */
 		private boolean observationsEnabled;
 
+		/**
+		 * Transaction settings.
+		 */
+		private final Transaction transaction = new TemplateTransaction();
+
 		public boolean isObservationsEnabled() {
 			return this.observationsEnabled;
 		}
@@ -866,6 +889,89 @@ public void setObservationsEnabled(boolean observationsEnabled) {
 			this.observationsEnabled = observationsEnabled;
 		}
 
+		public Transaction getTransaction() {
+			return this.transaction;
+		}
+
+	}
+
+	public abstract static class Transaction {
+
+		/**
+		 * Whether transactions are enabled for the component.
+		 */
+		private boolean enabled;
+
+		/**
+		 * Whether the component requires transactions.
+		 */
+		private boolean required;
+
+		/**
+		 * Duration representing the transaction timeout - null to use default timeout of
+		 * the underlying transaction system, or none if timeouts are not supported.
+		 */
+		private Duration timeout;
+
+		public boolean isEnabled() {
+			return this.enabled;
+		}
+
+		public void setEnabled(boolean enabled) {
+			this.enabled = enabled;
+		}
+
+		public boolean isRequired() {
+			return this.required;
+		}
+
+		public void setRequired(boolean required) {
+			this.required = required;
+		}
+
+		public Duration getTimeout() {
+			return this.timeout;
+		}
+
+		public void setTimeout(Duration timeout) {
+			this.timeout = timeout;
+		}
+
+		void validate() {
+			if (this.required && !this.enabled) {
+				String requiredProp = "%s.required".formatted(this.propertyPath());
+				String enabledProp = "%s.enabled".formatted(this.propertyPath());
+				throw new InvalidConfigurationPropertyValueException(requiredProp, this.required,
+						"Transactions must be enabled in order to be required. "
+								+ "Either set %s to 'true' or make transactions optional by setting %s to 'false'"
+									.formatted(enabledProp, requiredProp));
+			}
+		}
+
+		/**
+		 * Gets the property path that the transaction properties are mapped to.
+		 * @return the property path that the transaction properties are mapped to
+		 */
+		protected abstract String propertyPath();
+
+	}
+
+	static class TemplateTransaction extends Transaction {
+
+		@Override
+		protected String propertyPath() {
+			return "spring.pulsar.template.transaction";
+		}
+
+	}
+
+	static class ListenerTransaction extends Transaction {
+
+		@Override
+		protected String propertyPath() {
+			return "spring.pulsar.listener.transaction";
+		}
+
 	}
 
 	public static class Authentication {

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesMapper.java

@@ -39,6 +39,7 @@
 import org.apache.pulsar.client.impl.AutoClusterFailover.AutoClusterFailoverBuilderImpl;
 
 import org.springframework.boot.context.properties.PropertyMapper;
+import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
 import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
 import org.springframework.util.StringUtils;
@@ -64,6 +65,9 @@ void customizeClientBuilder(ClientBuilder clientBuilder, PulsarConnectionDetails
 		map.from(properties::getConnectionTimeout).to(timeoutProperty(clientBuilder::connectionTimeout));
 		map.from(properties::getOperationTimeout).to(timeoutProperty(clientBuilder::operationTimeout));
 		map.from(properties::getLookupTimeout).to(timeoutProperty(clientBuilder::lookupTimeout));
+		if (this.properties.isTransactionEnabled()) {
+			clientBuilder.enableTransaction(true);
+		}
 		customizeAuthentication(properties.getAuthentication(), clientBuilder::authentication);
 		customizeServiceUrlProviderBuilder(clientBuilder::serviceUrl, clientBuilder::serviceUrlProvider, properties,
 				connectionDetails);
@@ -157,6 +161,15 @@ <T> void customizeProducerBuilder(ProducerBuilder<T> producerBuilder) {
 		map.from(properties::getAccessMode).to(producerBuilder::accessMode);
 	}
 
+	<T> void customizeTemplate(PulsarTemplate<T> template) {
+		PulsarProperties.Transaction properties = this.properties.getTemplate().getTransaction();
+		properties.validate();
+		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		map.from(properties::isEnabled).to(template.transactions()::setEnabled);
+		map.from(properties::isRequired).to(template.transactions()::setRequired);
+		map.from(properties::getTimeout).to(template.transactions()::setTimeout);
+	}
+
 	<T> void customizeConsumerBuilder(ConsumerBuilder<T> consumerBuilder) {
 		PulsarProperties.Consumer properties = this.properties.getConsumer();
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
@@ -183,6 +196,7 @@ private void customizeConsumerBuilderSubscription(ConsumerBuilder<?> consumerBui
 	void customizeContainerProperties(PulsarContainerProperties containerProperties) {
 		customizePulsarContainerConsumerSubscriptionProperties(containerProperties);
 		customizePulsarContainerListenerProperties(containerProperties);
+		customizePulsarContainerTransactionProperties(containerProperties);
 	}
 
 	private void customizePulsarContainerConsumerSubscriptionProperties(PulsarContainerProperties containerProperties) {
@@ -198,6 +212,15 @@ private void customizePulsarContainerListenerProperties(PulsarContainerPropertie
 		map.from(properties::isObservationEnabled).to(containerProperties::setObservationEnabled);
 	}
 
+	private void customizePulsarContainerTransactionProperties(PulsarContainerProperties containerProperties) {
+		PulsarProperties.Transaction properties = this.properties.getListener().getTransaction();
+		properties.validate();
+		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		map.from(properties::isEnabled).to(containerProperties.transactions()::setEnabled);
+		map.from(properties::isRequired).to(containerProperties.transactions()::setRequired);
+		map.from(properties::getTimeout).to(containerProperties.transactions()::setTimeout);
+	}
+
 	<T> void customizeReaderBuilder(ReaderBuilder<T> readerBuilder) {
 		PulsarProperties.Reader properties = this.properties.getReader();
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();