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