GH-3439: Revise error handling in RedisStreamMP

Fixes #3439

The latest Spring Data Redis has introduced an `onErrorResume` function
option for the `StreamReceiver` and this one is now recommended way
to handle errors in the `Flux` from this receiver

* Expose all the `StreamReceiver.StreamReceiverOptionsBuilder` option
onto the `ReactiveRedisStreamMessageProducer`, including `onErrorResume`
* Have a default function as it was before - send into an error channel
supporting (n)ack in the failed message based on the failed record
* Make new setters mutually exclusive with an explicit `StreamReceiver.StreamReceiverOptions`

* Doc polishing

Author: artembilan

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,14 +17,19 @@
 package org.springframework.integration.redis.inbound;
 
 import java.time.Duration;
+import java.util.function.Function;
 
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.convert.ConversionFailedException;
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.ReadOffset;
 import org.springframework.data.redis.connection.stream.Record;
 import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.core.ReactiveRedisTemplate;
 import org.springframework.data.redis.core.ReactiveStreamOperations;
+import org.springframework.data.redis.hash.HashMapper;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.data.redis.stream.StreamReceiver;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -34,6 +39,7 @@
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.util.Assert;
@@ -61,12 +67,14 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
 
 	private final String streamKey;
 
-	private ReactiveStreamOperations<String, ?, ?> reactiveStreamOperations;
-
-	private StreamReceiver.StreamReceiverOptions<String, ?> streamReceiverOptions =
+	private final StreamReceiver.StreamReceiverOptionsBuilder<String, ?> streamReceiverOptionsBuilder =
 			StreamReceiver.StreamReceiverOptions.builder()
 					.pollTimeout(Duration.ZERO)
-					.build();
+					.onErrorResume(this::handleReceiverError);
+
+	private ReactiveStreamOperations<String, ?, ?> reactiveStreamOperations;
+
+	private StreamReceiver.StreamReceiverOptions<String, ?> streamReceiverOptions;
 
 	private StreamReceiver<String, ?> streamReceiver;
 
@@ -84,6 +92,8 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
 
 	private boolean createConsumerGroup;
 
+	private boolean receiverBuilderOptionSet;
+
 	public ReactiveRedisStreamMessageProducer(ReactiveRedisConnectionFactory reactiveConnectionFactory,
 			String streamKey) {
 
@@ -152,14 +162,105 @@ public void setCreateConsumerGroup(boolean createConsumerGroup) {
 	 * It provides a way to set the polling timeout and the serialization context.
 	 * By default the polling timeout is set to infinite and
 	 * {@link org.springframework.data.redis.serializer.StringRedisSerializer} is used.
+	 * Mutually exclusive with 'pollTimeout', 'batchSize', 'onErrorResume', 'serializer', 'targetType', 'objectMapper'.
 	 * @param streamReceiverOptions the desired receiver options
 	 * */
 	public void setStreamReceiverOptions(
 			@Nullable StreamReceiver.StreamReceiverOptions<String, ?> streamReceiverOptions) {
 
+		Assert.isTrue(!this.receiverBuilderOptionSet,
+				"The 'streamReceiverOptions' is mutually exclusive with 'pollTimeout', 'batchSize', " +
+						"'onErrorResume', 'serializer', 'targetType', 'objectMapper'");
 		this.streamReceiverOptions = streamReceiverOptions;
 	}
 
+	private void assertStreamReceiverOptions(String property) {
+		Assert.isNull(this.streamReceiverOptions,
+				() -> "'" + property + "' cannot be set when 'StreamReceiver.StreamReceiverOptions' is provided.");
+	}
+
+	/**
+	 * Configure a poll timeout for the BLOCK option during reading.
+	 * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}.
+	 * @param pollTimeout the timeout for polling.
+	 * @since 5.5
+	 * @see org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptionsBuilder#pollTimeout(Duration)
+	 */
+	public void setPollTimeout(Duration pollTimeout) {
+		assertStreamReceiverOptions("pollTimeout");
+		this.streamReceiverOptionsBuilder.pollTimeout(pollTimeout);
+		this.receiverBuilderOptionSet = true;
+	}
+
+	/**
+	 * Configure a batch size for the COUNT option during reading.
+	 * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}.
+	 * @param recordsPerPoll must be greater zero.
+	 * @since 5.5
+	 * @see org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptionsBuilder#batchSize(int)
+	 */
+	public void setBatchSize(int recordsPerPoll) {
+		assertStreamReceiverOptions("batchSize");
+		this.streamReceiverOptionsBuilder.batchSize(recordsPerPoll);
+		this.receiverBuilderOptionSet = true;
+	}
+
+	/**
+	 * Configure a resume Function to resume the main sequence when polling the stream fails.
+	 * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}.
+	 * By default this function extract the failed {@link Record} and sends an
+	 * {@link org.springframework.messaging.support.ErrorMessage} to the provided {@link #setErrorChannel(MessageChannel)}.
+	 * The failed message for this record may have a {@link IntegrationMessageHeaderAccessor#ACKNOWLEDGMENT_CALLBACK}
+	 * header when manual acknowledgment is configured for this message producer.
+	 * @param resumeFunction must not be null.
+	 * @since 5.5
+	 * @see org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptionsBuilder#onErrorResume(Function)
+	 */
+	public void setOnErrorResume(Function<? super Throwable, ? extends Publisher<Void>> resumeFunction) {
+		assertStreamReceiverOptions("onErrorResume");
+		this.streamReceiverOptionsBuilder.onErrorResume(resumeFunction);
+		this.receiverBuilderOptionSet = true;
+	}
+
+	/**
+	 * Configure a key, hash key and hash value serializer.
+	 * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}.
+	 * @param pair must not be null.
+	 * @since 5.5
+	 * @see StreamReceiver.StreamReceiverOptionsBuilder#serializer(RedisSerializationContext)
+	 */
+	public void setSerializer(RedisSerializationContext.SerializationPair<?> pair) {
+		assertStreamReceiverOptions("serializer");
+		this.streamReceiverOptionsBuilder.serializer(pair);
+		this.receiverBuilderOptionSet = true;
+	}
+
+	/**
+	 * Configure a hash target type. Changes the emitted Record type to ObjectRecord.
+	 * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}.
+	 * @param targetType must not be null.
+	 * @since 5.5
+	 * @see StreamReceiver.StreamReceiverOptionsBuilder#targetType(Class)
+	 */
+	public void setTargetType(Class<?> targetType) {
+		assertStreamReceiverOptions("targetType");
+		this.streamReceiverOptionsBuilder.targetType(targetType);
+		this.receiverBuilderOptionSet = true;
+	}
+
+	/**
+	 * Configure a hash mapper.
+	 * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}.
+	 * @param hashMapper must not be null.
+	 * @since 5.5
+	 * @see StreamReceiver.StreamReceiverOptionsBuilder#objectMapper(HashMapper)
+	 */
+	public void setObjectMapper(HashMapper<?, ?, ?> hashMapper) {
+		assertStreamReceiverOptions("objectMapper");
+		this.streamReceiverOptionsBuilder.objectMapper(hashMapper);
+		this.receiverBuilderOptionSet = true;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "redis:stream-inbound-channel-adapter";
@@ -168,6 +269,9 @@ public String getComponentType() {
 	@Override
 	protected void onInit() {
 		super.onInit();
+		if (this.streamReceiverOptions == null) {
+			this.streamReceiverOptions = this.streamReceiverOptionsBuilder.build();
+		}
 		this.streamReceiver = StreamReceiver.create(this.reactiveConnectionFactory, this.streamReceiverOptions);
 		if (StringUtils.hasText(this.consumerName) && !StringUtils.hasText(this.consumerGroup)) {
 			this.consumerGroup = getBeanName();
@@ -211,17 +315,7 @@ protected void doStart() {
 		}
 
 		Flux<? extends Message<?>> messageFlux =
-				events.map((record) -> buildMessageFromRecord(record, this.extractPayload))
-						.onErrorContinue((ex, record) -> {
-							@SuppressWarnings("unchecked")
-							Message<?> failedMessage = buildMessageFromRecord((Record<String, ?>) record, false);
-							MessagingException conversionException =
-									new MessageConversionException(failedMessage,
-											"Cannot deserialize Redis Stream Record", ex);
-							if (!sendErrorMessageIfNecessary(null, conversionException)) {
-								logger.getLog().error(conversionException);
-							}
-						});
+				events.map((record) -> buildMessageFromRecord(record, this.extractPayload));
 		subscribeToPublisher(messageFlux);
 	}
 
@@ -245,4 +339,22 @@ private Message<?> buildMessageFromRecord(Record<String, ?> record, boolean extr
 		return builder.build();
 	}
 
+	private <T> Publisher<T> handleReceiverError(Throwable error) {
+		Message<?> failedMessage = null;
+		if (error instanceof ConversionFailedException) {
+			@SuppressWarnings("unchecked")
+			Record<String, ?> record = (Record<String, ?>) ((ConversionFailedException) error).getValue();
+			if (record != null) {
+				failedMessage = buildMessageFromRecord(record, false);
+			}
+		}
+		MessagingException conversionException =
+				new MessageConversionException(failedMessage, // NOSONAR
+						"Cannot deserialize Redis Stream Record", error);
+		if (!sendErrorMessageIfNecessary(null, conversionException)) {
+			logger.getLog().error(conversionException);
+		}
+		return Mono.empty();
+	}
+
 }

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -327,11 +327,8 @@ PollableChannel redisStreamErrorChannel() {
 		ReactiveRedisStreamMessageProducer reactiveErrorRedisStreamProducer() {
 			ReactiveRedisStreamMessageProducer messageProducer =
 					new ReactiveRedisStreamMessageProducer(RedisAvailableRule.connectionFactory, STREAM_KEY);
-			messageProducer.setStreamReceiverOptions(
-					StreamReceiver.StreamReceiverOptions.builder()
-							.pollTimeout(Duration.ofMillis(100))
-							.targetType(Date.class)
-							.build());
+			messageProducer.setTargetType(Date.class);
+			messageProducer.setPollTimeout(Duration.ofMillis(100));
 			messageProducer.setCreateConsumerGroup(true);
 			messageProducer.setAutoAck(false);
 			messageProducer.setConsumerName("testConsumer");

src/reference/asciidoc/redis.adoc

@@ -1,4 +1,4 @@
-[[redis]]
+``[[redis]]
 == Redis Support
 
 Spring Integration 2.1 introduced support for https://redis.io/[Redis]: "`an open source advanced key-value store`".
@@ -859,6 +859,10 @@ Similar logic is required even when an exception happens during deserialization
 So, target error handler must decided to ack or nack such a failed message.
 Alongside with `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK`, the `ReactiveRedisStreamMessageProducer` also populates these headers into the message to produce: `RedisHeaders.STREAM_KEY`, `RedisHeaders.STREAM_MESSAGE_ID`, `RedisHeaders.CONSUMER_GROUP` and `RedisHeaders.CONSUMER`.
 
+Starting with version 5.5, you can configure `StreamReceiver.StreamReceiverOptionsBuilder` options explicitly on the `ReactiveRedisStreamMessageProducer`, including the newly introduced `onErrorResume` function, which is required if the Redis Stream consumer should continue polling when deserialization errors occur.
+The default function sends a message to the error channel (if provided) with possible acknowledgement for the failed message as it is described above.
+All these `StreamReceiver.StreamReceiverOptionsBuilder` are mutually exclusive with an externally provided `StreamReceiver.StreamReceiverOptions`.
+
 [[redis-lock-registry]]
 === Redis Lock Registry
 

src/reference/asciidoc/whats-new.adoc

@@ -23,3 +23,9 @@ If you are interested in more details, see the Issue Tracker tickets that were r
 
 The `AmqpInboundChannelAdapter` and `AmqpInboundGateway` (and the respective Java DSL builders) now support an `org.springframework.amqp.rabbit.retry.MessageRecoverer` as an AMQP-specific alternative to the general purpose `RecoveryCallback`.
 See <<./amqp.adoc#amqp,AMQP Support>> for more information.
+
+[[x5.5-redis]]
+==== Redis Changes
+
+The `ReactiveRedisStreamMessageProducer` has now setters for all the `StreamReceiver.StreamReceiverOptionsBuilder` options, including an `onErrorResume` function.
+See <<./redis.adoc#redis,Redis Support>> for more information.