Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3439: Revise error handling in RedisStreamMP

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/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
This commit is contained in:
Artem Bilan
2021-01-20 12:22:24 -05:00
committed by Gary Russell
parent c003a47379
commit 51e240b761
4 changed files with 142 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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,7 +17,11 @@
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;
@@ -25,6 +29,7 @@ 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.redis.support.RedisHeaders;
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 class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
* 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 class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
@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 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
}
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 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
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();
}
}

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

@@ -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 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
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");

6
src/reference/asciidoc/redis.adoc
View File

@@ -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

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -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.