From f08bfba70f815cfe42f66afc244770e3562cac10 Mon Sep 17 00:00:00 2001
From: Soby Chacko <soby.chacko@broadcom.com>
Date: Tue, 22 Sep 2026 16:34:09 -0400
Subject: [PATCH] Keep reader container running when listener throws

InternalAsyncReader only caught PulsarClientException, so any
exception thrown by the reader listener (e.g. a failing @PulsarReader
handler) escaped the read loop and silently terminated the reader
thread while the container still reported itself as running.

Catch and log listener exceptions so the reader continues with the
next message.

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
---
 .../DefaultPulsarMessageReaderContainer.java  |  4 +++
 ...aultPulsarMessageReaderContainerTests.java | 30 +++++++++++++++++++
 2 files changed, 34 insertions(+)

diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java
index 62c84d7c2..4c6e2ddce 100644
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java
@@ -231,6 +231,10 @@ public void run() {
 				catch (PulsarClientException e) {
 					DefaultPulsarMessageReaderContainer.this.logger.error(e, () -> "Error receiving messages.");
 				}
+				catch (Exception e) {
+					DefaultPulsarMessageReaderContainer.this.logger.error(e,
+							() -> "Error processing message in reader listener.");
+				}
 			}
 		}
 
diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java
index e20880c40..a442b59cb 100644
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java
@@ -182,6 +182,36 @@ void latestMessageId() throws Exception {
 		}
 	}
 
+	@Test
+	void readerContinuesAfterListenerThrowsException() throws Exception {
+		var latch = new CountDownLatch(1);
+		var containerProps = new PulsarReaderContainerProperties();
+		containerProps.setReaderListener((ReaderListener<?>) (reader, msg) -> {
+			if ("fail".equals(msg.getValue())) {
+				throw new IllegalStateException("Listener failed for " + msg.getValue());
+			}
+			latch.countDown();
+		});
+		containerProps.setStartMessageId(MessageId.earliest);
+		containerProps.setTopics(List.of("dprlct-004"));
+		containerProps.setSchema(Schema.STRING);
+
+		var readerFactory = new DefaultPulsarReaderFactory<String>(pulsarClient);
+		var container = new DefaultPulsarMessageReaderContainer<>(readerFactory, containerProps);
+		try {
+			container.start();
+			var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, "dprlct-004",
+					List.of((pb) -> pb.topic("dprlct-004")));
+			var pulsarTemplate = new PulsarTemplate<>(producerFactory);
+			pulsarTemplate.send("fail");
+			pulsarTemplate.send("hello after failure");
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+		finally {
+			safeStopContainer(container);
+		}
+	}
+
 	private void safeStopContainer(PulsarMessageReaderContainer container) {
 		try {
 			container.stop();