diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
index 3d34e09a..fc690ac7 100644
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
@@ -565,7 +565,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 					}
 				});
 			}
-			catch (RuntimeException ex) {
+			catch (Throwable ex) {
 				DefaultPulsarMessageListenerContainer.this.logger.error(ex, "Transaction rolled back");
 			}
 		}
@@ -651,7 +651,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 				return this.transactionTemplate.execute(status -> doInvokeBatchListener(messages, messageList,
 						inRetryMode, messagesPendingInBatch, getTransaction()));
 			}
-			catch (RuntimeException e) {
+			catch (Throwable e) {
 				DefaultPulsarMessageListenerContainer.this.logger.error(e, "Transaction rolled back");
 				return Collections.emptyList();
 			}
diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java
index 9114a33f..2c30e440 100644
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java
@@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -39,6 +40,7 @@ import org.testcontainers.junit.jupiter.Testcontainers;
 
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.ProducerBuilderCustomizer;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.test.support.PulsarConsumerTestUtil;
 import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
@@ -58,7 +60,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 
 	private PulsarClient client;
 
-	private PulsarTemplate<String> pulsarTemplate;
+	private PulsarTemplate<String> transactionalPulsarTemplate;
 
 	private PulsarTransactionManager transactionManager;
 
@@ -74,7 +76,8 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 			.serviceUrl(PULSAR_CONTAINER.getPulsarBrokerUrl())
 			.build();
 		var producerFactory = new DefaultPulsarProducerFactory<String>(client);
-		pulsarTemplate = new PulsarTemplate<>(producerFactory);
+		transactionalPulsarTemplate = new PulsarTemplate<>(producerFactory);
+		transactionalPulsarTemplate.transactions().setEnabled(true);
 		transactionManager = new PulsarTransactionManager(client);
 	}
 
@@ -90,8 +93,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var containerProps = newContainerProps();
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+			transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 			listenerLatch.countDown();
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, false, "msg1");
@@ -105,8 +107,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var containerProps = newContainerProps();
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+			transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 			listenerLatch.countDown();
 		});
@@ -122,8 +123,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		containerProps.setAckMode(AckMode.MANUAL);
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarAcknowledgingMessageListener<?>) (consumer, msg, ack) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+			transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 			ack.acknowledge();
 			listenerLatch.countDown();
 		});
@@ -139,8 +139,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		containerProps.setAckMode(AckMode.MANUAL);
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarAcknowledgingMessageListener<?>) (consumer, msg, ack) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+			transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 			ack.acknowledge();
 			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 			listenerLatch.countDown();
@@ -156,8 +155,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var containerProps = newContainerProps();
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+			transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 			listenerLatch.countDown();
 			throw new RuntimeException("BOOM");
 		});
@@ -172,8 +170,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var containerProps = newContainerProps();
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out"));
+			transactionalPulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out"));
 			listenerLatch.countDown();
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, false, "msg1");
@@ -187,8 +184,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var containerProps = newContainerProps();
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out"));
+			transactionalPulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out"));
 			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 			listenerLatch.countDown();
 		});
@@ -204,8 +200,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
 		var listenerLatch = new CountDownLatch(inputMsgs.size());
 		containerProps.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+			transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 			listenerLatch.countDown();
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, false, inputMsgs);
@@ -221,8 +216,7 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
 		var listenerLatch = new CountDownLatch(inputMsgs.size());
 		containerProps.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			pulsarTemplate.transactions().setEnabled(true);
-			pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+			transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 			listenerLatch.countDown();
 			if (msg.getValue().equals("msg2")) {
 				throw new RuntimeException("BOOM-msg2");
@@ -255,12 +249,12 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		containerProps.setAckMode(AckMode.BATCH);
 		containerProps.setSubscriptionType(SubscriptionType.Shared);
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
-		var listenerLatch = new CountDownLatch(1);
+		var listenerLatch = new CountDownLatch(inputMsgs.size());
 		containerProps.setMessageListener((PulsarBatchMessageListener<?>) (consumer, msgs) -> {
-			assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-			pulsarTemplate.transactions().setEnabled(true);
-			msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out"));
-			listenerLatch.countDown();
+			msgs.forEach((msg) -> {
+				transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
+				listenerLatch.countDown();
+			});
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs);
 		var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList();
@@ -278,12 +272,12 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		containerProps.setAckMode(AckMode.BATCH);
 		containerProps.setSubscriptionType(SubscriptionType.Exclusive);
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
-		var listenerLatch = new CountDownLatch(1);
+		var listenerLatch = new CountDownLatch(inputMsgs.size());
 		containerProps.setMessageListener((PulsarBatchMessageListener<?>) (consumer, msgs) -> {
-			assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-			pulsarTemplate.transactions().setEnabled(true);
-			msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out"));
-			listenerLatch.countDown();
+			msgs.forEach((msg) -> {
+				transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
+				listenerLatch.countDown();
+			});
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs);
 		var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList();
@@ -302,10 +296,8 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarBatchMessageListener<?>) (consumer, msgs) -> {
-			assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-			pulsarTemplate.transactions().setEnabled(true);
-			msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out"));
-			listenerLatch.countDown();
+			msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"));
+			CompletableFuture.runAsync(() -> listenerLatch.countDown());
 			throw new RuntimeException("NOPE");
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs);
@@ -323,11 +315,9 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarBatchMessageListener<?>) (consumer, msgs) -> {
-			assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-			pulsarTemplate.transactions().setEnabled(true);
-			msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out"));
-			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
+			msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"));
 			listenerLatch.countDown();
+			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs);
 		assertNoMessagesAvailableInOutputTopic(topicOut);
@@ -343,22 +333,20 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarBatchMessageListener<?>) (consumer, msgs) -> {
-			assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-			pulsarTemplate.transactions().setEnabled(true);
 			msgs.forEach((msg) -> {
-				if (msg.getValue().equals("msg2")) {
-					pulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out"));
+				if (msg.getValue().equals("msg1")) {
+					transactionalPulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out"));
 				}
 				else {
-					pulsarTemplate.send(topicOut, msg.getValue() + "-out");
+					transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out");
 				}
 			});
-			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 			listenerLatch.countDown();
+			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs);
-		// msg1 and msg2 get rollback but nested txn for msg2 gets committed
-		assertMessagesAvailableInOutputTopic(topicOut, "msg2-out");
+		// msg2 and msg3 get rollback but nested txn for msg1 gets committed
+		assertMessagesAvailableInOutputTopic(topicOut, "msg1-out");
 	}
 
 	@Test
@@ -369,13 +357,11 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		containerProps.setBatchListener(true);
 		containerProps.setAckMode(AckMode.MANUAL);
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
-		var listenerLatch = new CountDownLatch(1);
+		var listenerLatch = new CountDownLatch(inputMsgs.size());
 		containerProps.setMessageListener((PulsarBatchAcknowledgingMessageListener<?>) (consumer, msgs, ack) -> {
-			assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-			pulsarTemplate.transactions().setEnabled(true);
-			msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out"));
+			msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"));
 			ack.acknowledge(msgs.stream().map(Message::getMessageId).toList());
-			listenerLatch.countDown();
+			msgs.forEach((__) -> listenerLatch.countDown());
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs);
 		var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList();
@@ -392,12 +378,10 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var inputMsgs = List.of("msg1", "msg2", "msg3");
 		var listenerLatch = new CountDownLatch(1);
 		containerProps.setMessageListener((PulsarBatchAcknowledgingMessageListener<?>) (consumer, msgs, ack) -> {
-			assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-			pulsarTemplate.transactions().setEnabled(true);
-			msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out"));
+			msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"));
 			ack.acknowledge(msgs.stream().map(Message::getMessageId).toList());
-			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 			listenerLatch.countDown();
+			PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly();
 		});
 		startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs);
 		assertNoMessagesAvailableInOutputTopic(topicOut);
@@ -423,25 +407,29 @@ class DefaultPulsarMessageListenerContainerTxnTests {
 		var container = new DefaultPulsarMessageListenerContainer<>(consumerFactory, containerProps);
 		try {
 			container.start();
-			pulsarTemplate.transactions().setEnabled(false);
-			if (sendInBatch) {
-				inputMsgs.forEach((msg) -> pulsarTemplate.newMessage(msg)
-					.withTopic(topicIn)
-					.withProducerCustomizer((pb) -> pb.enableBatching(true)
-						.batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS)
-						.batchingMaxMessages(inputMsgs.size()))
-					.sendAsync());
-			}
-			else {
-				inputMsgs.forEach((msg) -> pulsarTemplate.sendAsync(topicIn, msg));
-			}
+			var nonTransactionalTemplate = newNonTransactionalTemplate(sendInBatch, inputMsgs.size());
+			inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg));
 			assertThat(listenerLatch.await(sendInBatch ? 8 : 5, TimeUnit.SECONDS)).isTrue();
+			if (sendInBatch) {
+				// Because the latch may fire before exception is thrown - give it a pause
+				Thread.sleep(500);
+			}
 		}
 		finally {
 			container.stop();
 		}
 	}
 
+	private PulsarTemplate<String> newNonTransactionalTemplate(boolean sendInBatch, int numInBatch) {
+		List<ProducerBuilderCustomizer<String>> customizers = List.of();
+		if (sendInBatch) {
+			customizers = List.of((pb) -> pb.enableBatching(true)
+				.batchingMaxPublishDelay(2, TimeUnit.SECONDS)
+				.batchingMaxMessages(numInBatch));
+		}
+		return new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(client, null, customizers));
+	}
+
 	private void assertNoMessagesAvailableInOutputTopic(String topicOut) {
 		assertThat(PulsarConsumerTestUtil.<String>consumeMessages(client)
 			.fromTopic(topicOut)
diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java
index 97ffa866..32e04947 100644
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java
@@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -33,6 +34,8 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarListener;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.ProducerBuilderCustomizer;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.listener.PulsarListenerTxnTests.BatchListenerWithCommit.BatchListenerWithCommitConfig;
 import org.springframework.pulsar.listener.PulsarListenerTxnTests.BatchListenerWithRollback.BatchListenerWithRollbackConfig;
@@ -51,10 +54,6 @@ import org.springframework.transaction.annotation.Transactional;
  */
 class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 
-	private void sendInputMessageNonTransactionally(String topic, String msg) {
-		nonTransactionalPulsarTemplate.send(topic, msg);
-	}
-
 	private void assertNoMessagesAvailableInOutputTopic(String topicOut) {
 		assertThat(PulsarConsumerTestUtil.<String>consumeMessages(pulsarClient)
 			.fromTopic(topicOut)
@@ -75,6 +74,16 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 			.get()).map(Message::getValue).containsExactlyInAnyOrderElementsOf(expectedMessages);
 	}
 
+	private PulsarTemplate<String> newNonTransactionalTemplate(boolean sendInBatch, int numMessages) {
+		List<ProducerBuilderCustomizer<String>> customizers = List.of();
+		if (sendInBatch) {
+			customizers = List.of((pb) -> pb.enableBatching(true)
+				.batchingMaxPublishDelay(2, TimeUnit.SECONDS)
+				.batchingMaxMessages(numMessages));
+		}
+		return new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient, null, customizers));
+	}
+
 	@Nested
 	@ContextConfiguration(classes = ListenerWithExternalTransactionConfig.class)
 	class ListenerWithExternalTransaction {
@@ -85,7 +94,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 
 		@Test
 		void producedMessageIsCommitted() throws Exception {
-			sendInputMessageNonTransactionally(topicIn, "msg1");
+			var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1);
+			nonTransactionalTemplate.send(topicIn, "msg1");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 			assertMessagesAvailableInOutputTopic(topicOut, "msg1-out");
 		}
@@ -118,7 +128,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 
 		@Test
 		void producedMessageIsNotCommitted() throws Exception {
-			sendInputMessageNonTransactionally(topicIn, "msg1");
+			var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1);
+			nonTransactionalTemplate.send(topicIn, "msg1");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 			assertNoMessagesAvailableInOutputTopic(topicOut);
 		}
@@ -152,7 +163,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 
 		@Test
 		void producedMessageIsCommitted() throws Exception {
-			sendInputMessageNonTransactionally(topicIn, "msg1");
+			var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1);
+			nonTransactionalTemplate.send(topicIn, "msg1");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 			assertMessagesAvailableInOutputTopic(topicOut, "msg1-out");
 		}
@@ -184,7 +196,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 
 		@Test
 		void producedMessageIsNotCommitted() throws Exception {
-			sendInputMessageNonTransactionally(topicIn, "msg1");
+			var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1);
+			nonTransactionalTemplate.send(topicIn, "msg1");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 			assertNoMessagesAvailableInOutputTopic(topicOut);
 		}
@@ -211,20 +224,16 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 	@ContextConfiguration(classes = BatchListenerWithCommitConfig.class)
 	class BatchListenerWithCommit {
 
-		static final CountDownLatch latch = new CountDownLatch(1);
 		static final String topicIn = "pltt-batch-lstnr-in";
 		static final String topicOut = "pltt-batch-lstnr-out";
 		static final List<String> inputMsgs = List.of("msg1", "msg2", "msg3");
+		static final CountDownLatch latch = new CountDownLatch(inputMsgs.size());
 
 		@Test
 		void producedMessagesAreCommitted() throws Exception {
-			inputMsgs.forEach((msg) -> nonTransactionalPulsarTemplate.newMessage(msg)
-				.withTopic(topicIn)
-				.withProducerCustomizer((pb) -> pb.enableBatching(true)
-					.batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS)
-					.batchingMaxMessages(inputMsgs.size()))
-				.sendAsync());
-			assertThat(latch.await(15, TimeUnit.SECONDS)).isTrue();
+			var nonTransactionalTemplate = newNonTransactionalTemplate(true, inputMsgs.size());
+			inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg));
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 			var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList();
 			assertMessagesAvailableInOutputTopic(topicOut, outputMsgs);
 		}
@@ -238,9 +247,10 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 
 			@PulsarListener(topics = topicIn, batch = true)
 			void listen(List<String> msgs) {
-				assertThat(msgs.size()).isEqualTo(inputMsgs.size());
-				msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg + "-out"));
-				latch.countDown();
+				msgs.forEach((msg) -> {
+					transactionalPulsarTemplate.send(topicOut, msg + "-out");
+					latch.countDown();
+				});
 			}
 
 		}
@@ -251,20 +261,16 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 	@ContextConfiguration(classes = BatchListenerWithRollbackConfig.class)
 	class BatchListenerWithRollback {
 
-		static final CountDownLatch latch = new CountDownLatch(1);
 		static final String topicIn = "pltt-batch-lstnr-rb-in";
 		static final String topicOut = "pltt-batch-lstnr-rb-out";
 		static final List<String> inputMsgs = List.of("msg1", "msg2", "msg3");
+		static final CountDownLatch latch = new CountDownLatch(1);
 
 		@Test
 		void producedMessagesAreNotCommitted() throws Exception {
-			inputMsgs.forEach((msg) -> nonTransactionalPulsarTemplate.newMessage(msg)
-				.withTopic(topicIn)
-				.withProducerCustomizer((pb) -> pb.enableBatching(true)
-					.batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS)
-					.batchingMaxMessages(inputMsgs.size()))
-				.sendAsync());
-			assertThat(latch.await(15, TimeUnit.SECONDS)).isTrue();
+			var nonTransactionalTemplate = newNonTransactionalTemplate(true, inputMsgs.size());
+			inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg));
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 			assertNoMessagesAvailableInOutputTopic(topicOut);
 		}
 
@@ -277,9 +283,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
 
 			@PulsarListener(topics = topicIn, batch = true)
 			void listen(List<String> msgs) {
-				assertThat(msgs.size()).isEqualTo(inputMsgs.size());
 				msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg + "-out"));
-				latch.countDown();
+				CompletableFuture.runAsync(() -> latch.countDown());
 				throw new RuntimeException("BOOM-batch");
 			}
 
diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java
index d82e7906..c1f30fdd 100644
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java
@@ -23,7 +23,6 @@ import org.testcontainers.junit.jupiter.Testcontainers;
 
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.pulsar.annotation.EnablePulsar;
@@ -65,13 +64,8 @@ class PulsarTxnTestsBase {
 	protected PulsarClient pulsarClient;
 
 	@Autowired
-	@Qualifier("transactionalPulsarTemplate")
 	protected PulsarTemplate<String> transactionalPulsarTemplate;
 
-	@Autowired
-	@Qualifier("nonTransactionalPulsarTemplate")
-	protected PulsarTemplate<String> nonTransactionalPulsarTemplate;
-
 	@Configuration(proxyBeanMethods = false)
 	@EnablePulsar
 	static class TopLevelConfig {
@@ -96,13 +90,6 @@ class PulsarTxnTestsBase {
 			return template;
 		}
 
-		@Bean
-		PulsarTemplate<String> nonTransactionalPulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
-			var template = new PulsarTemplate<>(pulsarProducerFactory);
-			template.transactions().setEnabled(false);
-			return template;
-		}
-
 		@Bean
 		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
 				ObjectProvider<ConsumerBuilderCustomizer<String>> defaultConsumerCustomizersProvider) {