WIP

Author: onobc

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java

@@ -73,7 +73,7 @@ public class ReactivePulsarContainerProperties<T> {
 		.fixedBackoff(Duration.ofSeconds(10))
 		.build();
 
-	private StartupFailurePolicy startupFailurePolicy = StartupFailurePolicy.CONTINUE;
+	private StartupFailurePolicy startupFailurePolicy = StartupFailurePolicy.STOP;
 
 	public ReactivePulsarMessageHandler getMessageHandler() {
 		return this.messageHandler;

spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java

@@ -108,7 +108,7 @@ public class PulsarContainerProperties {
 		.fixedBackoff(Duration.ofSeconds(10))
 		.build();
 
-	private StartupFailurePolicy startupFailurePolicy = StartupFailurePolicy.CONTINUE;
+	private StartupFailurePolicy startupFailurePolicy = StartupFailurePolicy.STOP;
 
 	public PulsarContainerProperties(String... topics) {
 		this.topics = Set.of(topics);

spring-pulsar/src/main/java/org/springframework/pulsar/reader/PulsarReaderContainerProperties.java

@@ -66,7 +66,7 @@ public class PulsarReaderContainerProperties {
 		.fixedBackoff(Duration.ofSeconds(10))
 		.build();
 
-	private StartupFailurePolicy startupFailurePolicy = StartupFailurePolicy.CONTINUE;
+	private StartupFailurePolicy startupFailurePolicy = StartupFailurePolicy.STOP;
 
 	public Object getReaderListener() {
 		return this.readerListener;

spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java

@@ -431,7 +431,9 @@ void batchListenerWithRecordAckModeNotSupported() {
 		var consumerFactory = new DefaultPulsarConsumerFactory<String>(mock(PulsarClient.class), List.of());
 		var container = new DefaultPulsarMessageListenerContainer<>(consumerFactory, containerProps);
 		assertThatIllegalStateException().isThrownBy(() -> container.start())
-			.withMessage("Transactional batch listeners do not support AckMode.RECORD");
+				.withCauseInstanceOf(IllegalStateException.class)
+				.havingRootCause()
+				.withMessage("Transactional batch listeners do not support AckMode.RECORD");
 	}
 
 	@Test

spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java

@@ -248,7 +248,9 @@ void recordListenerWithBatchAckNotSupported() {
 		var consumerFactory = new DefaultPulsarConsumerFactory<String>(client, List.of());
 		var container = new DefaultPulsarMessageListenerContainer<>(consumerFactory, containerProps);
 		assertThatIllegalStateException().isThrownBy(() -> container.start())
-			.withMessage("Transactional record listeners can not use batch ack mode");
+				.withCauseInstanceOf(IllegalStateException.class)
+				.havingRootCause()
+				.withMessage("Transactional record listeners can not use batch ack mode");
 	}
 
 	@Test
@@ -402,7 +404,7 @@ void txnBatchListenerWithErrorHandlerNotSupported() {
 		var container = new DefaultPulsarMessageListenerContainer<>(consumerFactory, containerProps);
 		container.setPulsarConsumerErrorHandler(mock(PulsarConsumerErrorHandler.class));
 		assertThatIllegalStateException().isThrownBy(() -> container.start())
-			.withMessage("Transactional batch listeners do not support custom error handlers");
+			.withStackTraceContaining("Transactional batch listeners do not support custom error handlers");
 	}
 
 	private Consumer<String> startContainerAndSendInputsThenWaitForLatch(String topicIn,

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java

@@ -322,7 +322,7 @@ void ignoresSettingWhenNoTxnManagerAvailable() {
 				context.refresh();
 			})
 				.withCauseInstanceOf(IllegalStateException.class)
-				.havingCause()
+				.havingRootCause()
 				.withMessage("Transactions are enabled but txn manager is not set");
 		}