Keep reader container running when listener throws

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.");
+				}
 			}
 		}
 

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();