Add Spring Pulsar transaction support

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

@@ -34,6 +34,7 @@
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.schema.SchemaType;
 
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 import org.springframework.util.Assert;
@@ -47,7 +48,7 @@
  * @since 3.2.0
  */
 @ConfigurationProperties("spring.pulsar")
-public class PulsarProperties {
+public class PulsarProperties implements InitializingBean {
 
 	private final Client client = new Client();
 
@@ -103,11 +104,17 @@ public Template getTemplate() {
 		return this.template;
 	}
 
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		this.getTemplate().getTransaction().validate("spring.pulsar.template");
+		this.getListener().getTransaction().validate("spring.pulsar.listener");
+	}
+
 	/**
 	 * Whether transactions are enabled for either the template or the listener.
 	 * @return whether transactions are enabled for either the template or the listener
 	 */
-	public boolean isTransactionEnabled() {
+	boolean isTransactionEnabled() {
 		return this.template.getTransaction().isEnabled() || this.listener.getTransaction().isEnabled();
 	}
 
@@ -897,7 +904,7 @@ public Transaction getTransaction() {
 	public static class Transaction {
 
 		/**
-		 * Whether the component supports transactions.
+		 * Whether transactions are enabled for the component.
 		 */
 		private boolean enabled;
 
@@ -936,6 +943,11 @@ public void setTimeout(Duration timeout) {
 			this.timeout = timeout;
 		}
 
+		void validate(String prefix) {
+			Assert.state((this.enabled || !this.required), "If transactions are required "
+					+ "they must also be enabled - consult your '%s.transaction' properties.".formatted(prefix));
+		}
+
 	}
 
 	public static class Authentication {

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

@@ -65,8 +65,7 @@ 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.getTemplate().getTransaction().isEnabled()
-				|| this.properties.getListener().getTransaction().isEnabled()) {
+		if (this.properties.isTransactionEnabled()) {
 			clientBuilder.enableTransaction(true);
 		}
 		customizeAuthentication(properties.getAuthentication(), clientBuilder::authentication);

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarAutoConfigurationTests.java

@@ -665,6 +665,28 @@ void whenTemplateEnablesTransactionsAutoconfiguresBean() {
 				.run((context) -> assertThat(context).hasSingleBean(PulsarAwareTransactionManager.class));
 		}
 
+		@Test
+		void whenTemplateRequiresTransactionsThenTransactionsMustBeEnabled() {
+			this.contextRunner
+				.withPropertyValues("spring.pulsar.template.transaction.required=true",
+						"spring.pulsar.template.transaction.enabled=false")
+				.run((context) -> assertThat(context).hasFailed()
+					.getFailure()
+					.hasMessageEndingWith(
+							"If transactions are required they must also be enabled - consult your 'spring.pulsar.template.transaction' properties."));
+		}
+
+		@Test
+		void whenListenerRequiresTransactionsThenTransactionsMustBeEnabled() {
+			this.contextRunner
+				.withPropertyValues("spring.pulsar.listener.transaction.required=true",
+						"spring.pulsar.listener.transaction.enabled=false")
+				.run((context) -> assertThat(context).hasFailed()
+					.getFailure()
+					.hasMessageEndingWith(
+							"If transactions are required they must also be enabled - consult your 'spring.pulsar.listener.transaction' properties."));
+		}
+
 	}
 
 }

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesTests.java

@@ -415,19 +415,39 @@ void bind() {
 	@Nested
 	class TransactionProperties {
 
-		@ParameterizedTest
-		@MethodSource
-		void transactionsEnabledTests(boolean listenerEnablesTransactions, boolean templateEnablesTransactions,
-				boolean shouldTransactionsBeEnabled) {
+		@Test
+		void transactionsEnabledWhenListenerAndTemplateBothEnabled() {
+			PulsarProperties properties = new PulsarProperties();
+			properties.getListener().getTransaction().setEnabled(true);
+			properties.getTemplate().getTransaction().setEnabled(true);
+			assertThat(properties.isTransactionEnabled()).isTrue();
+
+		}
+
+		@Test
+		void transactionsEnabledWhenListenerEnabledAndTemplateDisabled() {
+			PulsarProperties properties = new PulsarProperties();
+			properties.getListener().getTransaction().setEnabled(true);
+			properties.getTemplate().getTransaction().setEnabled(false);
+			assertThat(properties.isTransactionEnabled()).isTrue();
+
+		}
+
+		@Test
+		void transactionsEnabledWhenListenerDisabledAndTemplateEnabled() {
 			PulsarProperties properties = new PulsarProperties();
-			properties.getListener().getTransaction().setEnabled(listenerEnablesTransactions);
-			properties.getTemplate().getTransaction().setEnabled(templateEnablesTransactions);
-			assertThat(properties.isTransactionEnabled()).isEqualTo(shouldTransactionsBeEnabled);
+			properties.getListener().getTransaction().setEnabled(false);
+			properties.getTemplate().getTransaction().setEnabled(true);
+			assertThat(properties.isTransactionEnabled()).isTrue();
+
 		}
 
-		static Stream<Arguments> transactionsEnabledTests() {
-			return Stream.of(Arguments.arguments(true, true, true), Arguments.arguments(true, false, true),
-					Arguments.arguments(false, true, true), Arguments.arguments(false, false, false));
+		void transactionsDisabledWhenListenerAndTemplateBothDisabled() {
+			PulsarProperties properties = new PulsarProperties();
+			properties.getListener().getTransaction().setEnabled(false);
+			properties.getTemplate().getTransaction().setEnabled(false);
+			assertThat(properties.isTransactionEnabled()).isFalse();
+
 		}
 
 	}