Reactive StreamReceiver fails on serialization exception and does not process subsequent messages [DATAREDIS-1230]

[spring-projects-issues]

Rohan Mukesh opened DATAREDIS-1230 and commented

I am using Spring Integration Reactive Redis which internally uses StreamReceiver to receive messages from Redis stream. StreamReceiver is configured to use StreamOptions.
Whenever there is SerializationException for message in Stream, StreamReceiver stops receiving next messages and we have to restart the application to consume next messages on the stream.

Artem Bilan has provided a fix for Spring Integration to use onErrorContinue . It would be great for Spring Data Redis to extend the StreamReceiver to handle error

---

Referenced from: pull request #576

Backported to: 2.4.2 (2020.0.2)

[spring-projects-issues]

Mark Paluch commented

onErrorContinue happens a bit late in the receive process. We could introduce an error handling function similar to the imperative variant StreamMessageListenerContainer. StreamMessageListenerContainer defines ErrorHandler and a Predicate<Throwable> cancelSubscriptionOnError to control the exit condition.

The handling function would be a Function<Throwable, V> where V is defined as V extends Record<K, ?> (or Object instead of ? to actually allow producing that type).

Since the reactive model is stream-oriented, we likely don't require this split and can simplify things a bit

[spring-projects-issues]

Mark Paluch commented

A fix requires a bit more effort than onErrorResume as internally the read offset/polling state was not advanced on read errors. Also, if a batch of records is read and a single record fails to deserialize, then the imperative variant fails with the entire batch instead of emitting records until reaching the failure

[spring-projects-issues]

Artem Bilan commented

Thank you for fixing this!

We will revise the logic in Spring Integration for the next version: spring-projects/spring-integration#3439