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

Add Error Handling to RedisStreamMessageProducer

Browse Source 
* Use 'onErrorContinue' to continue receiving messages from Stream
* Send an `ErrorMessage` to the provided `errorChannel` (if any)
* And `@Nullable` to some `MessageProducerSupport` API which definitely
may accept `null`
* Extract common `buildMessageFromRecord()` method in the `ReactiveRedisStreamMessageProducer`,
so all the headers from the stream `Record` are carried to the message independently of the
 record state - normal send or error sending
This commit is contained in:
rohan mukesh
2020-10-13 13:33:17 -05:00
committed by Artem Bilan
parent 5b74db8417
commit b1a383060d
3 changed files with 151 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -248,7 +248,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
* @return true if the error channel is available and message sent.
* @since 4.3.10
*/
protected final boolean sendErrorMessageIfNecessary(Message<?> message, Exception exception) {
protected final boolean sendErrorMessageIfNecessary(@Nullable Message<?> message, Exception exception) {
MessageChannel channel = getErrorChannel();
if (channel != null) {
this.messagingTemplate.send(channel, buildErrorMessage(message, exception));
@@ -265,7 +265,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
* @return the error message.
* @since 4.3.10
*/
protected final ErrorMessage buildErrorMessage(Message<?> message, Exception exception) {
protected final ErrorMessage buildErrorMessage(@Nullable Message<?> message, Exception exception) {
return this.errorMessageStrategy.buildErrorMessage(exception, getErrorMessageAttributes(message));
}
@@ -277,7 +277,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
* @return the attributes.
* @since 4.3.10
*/
protected AttributeAccessor getErrorMessageAttributes(Message<?> message) {
protected AttributeAccessor getErrorMessageAttributes(@Nullable Message<?> message) {
return ErrorMessageUtils.getAttributeAccessor(message, null);
}

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

@@ -34,6 +34,8 @@ 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.MessagingException;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -211,25 +213,38 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
}
Flux<? extends Message<?>> messageFlux =
events.map((event) -> {
AbstractIntegrationMessageBuilder<?> builder =
getMessageBuilderFactory()
.withPayload(this.extractPayload ? event.getValue() : event)
.setHeader(RedisHeaders.STREAM_KEY, event.getStream())
.setHeader(RedisHeaders.STREAM_MESSAGE_ID, event.getId())
.setHeader(RedisHeaders.CONSUMER_GROUP, this.consumerGroup)
.setHeader(RedisHeaders.CONSUMER, this.consumerName);
if (!this.autoAck && this.consumerGroup != null) {
builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
(SimpleAcknowledgment) () ->
this.reactiveStreamOperations
.acknowledge(this.consumerGroup, event)
.subscribe());
}
return builder.build();
});
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);
}
});
subscribeToPublisher(messageFlux);
}
private Message<?> buildMessageFromRecord(Record<String, ?> record, boolean extractPayload) {
AbstractIntegrationMessageBuilder<?> builder =
getMessageBuilderFactory()
.withPayload(extractPayload ? record.getValue() : record)
.setHeader(RedisHeaders.STREAM_KEY, record.getStream())
.setHeader(RedisHeaders.STREAM_MESSAGE_ID, record.getId())
.setHeader(RedisHeaders.CONSUMER_GROUP, this.consumerGroup)
.setHeader(RedisHeaders.CONSUMER, this.consumerName);
if (!this.autoAck && this.consumerGroup != null) {
builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
(SimpleAcknowledgment) () ->
this.reactiveStreamOperations
.acknowledge(this.consumerGroup, record)
.subscribe());
}
return builder.build();
}
}

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

@@ -20,8 +20,10 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.atomic.AtomicReference;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -32,7 +34,6 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.stream.StreamReceiver;
@@ -40,6 +41,7 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.acks.SimpleAcknowledgment;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
import org.springframework.integration.redis.outbound.ReactiveRedisStreamMessageHandler;
import org.springframework.integration.redis.rules.RedisAvailable;
@@ -48,6 +50,10 @@ import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.redis.support.RedisHeaders;
import org.springframework.integration.redis.util.Address;
import org.springframework.integration.redis.util.Person;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -75,7 +81,7 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
FluxMessageChannel fluxMessageChannel;
@Autowired
ReactiveRedisStreamMessageProducer redisStreamMessageProducer;
ReactiveRedisStreamMessageProducer reactiveRedisStreamProducer;
@Autowired
ReactiveRedisTemplate<String, ?> template;
@@ -85,42 +91,32 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
@Before
public void delKey() {
this.template.hasKey(STREAM_KEY)
.filter(Boolean::booleanValue)
.flatMapMany(b ->
this.template.opsForStream()
.groups(STREAM_KEY)
.map(StreamInfo.XInfoGroup::groupName)
.flatMap(groupName ->
this.template.opsForStream()
.destroyGroup(STREAM_KEY, groupName)))
.blockLast();
this.template.delete(STREAM_KEY).block();
}
@After
public void tearDown() {
this.redisStreamMessageProducer.stop();
this.reactiveRedisStreamProducer.stop();
RedisAvailableRule.connectionFactory.resetConnection();
}
@Test
@RedisAvailable
public void testConsumerGroupCreation() {
this.redisStreamMessageProducer.setCreateConsumerGroup(true);
this.redisStreamMessageProducer.setConsumerName(CONSUMER);
this.redisStreamMessageProducer.afterPropertiesSet();
this.reactiveRedisStreamProducer.setCreateConsumerGroup(true);
this.reactiveRedisStreamProducer.setConsumerName(CONSUMER);
this.reactiveRedisStreamProducer.afterPropertiesSet();
Flux.from(this.fluxMessageChannel).subscribe();
this.redisStreamMessageProducer.start();
this.reactiveRedisStreamProducer.start();
this.template.opsForStream()
.groups(STREAM_KEY)
.next()
.as(StepVerifier::create)
.assertNext((infoGroup) ->
assertThat(infoGroup.groupName()).isEqualTo(this.redisStreamMessageProducer.getBeanName()))
assertThat(infoGroup.groupName()).isEqualTo(this.reactiveRedisStreamProducer.getBeanName()))
.thenCancel()
.verify(Duration.ofSeconds(10));
}
@@ -132,10 +128,10 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
Person person = new Person(address, "Attoumane");
this.messageHandler.handleMessage(new GenericMessage<>(person));
this.redisStreamMessageProducer.setCreateConsumerGroup(false);
this.redisStreamMessageProducer.setConsumerName(null);
this.redisStreamMessageProducer.setReadOffset(ReadOffset.from("0-0"));
this.redisStreamMessageProducer.afterPropertiesSet();
this.reactiveRedisStreamProducer.setCreateConsumerGroup(false);
this.reactiveRedisStreamProducer.setConsumerName(null);
this.reactiveRedisStreamProducer.setReadOffset(ReadOffset.from("0-0"));
this.reactiveRedisStreamProducer.afterPropertiesSet();
StepVerifier stepVerifier =
Flux.from(this.fluxMessageChannel)
@@ -148,7 +144,7 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
.thenCancel()
.verifyLater();
this.redisStreamMessageProducer.start();
this.reactiveRedisStreamProducer.start();
stepVerifier.verify(Duration.ofSeconds(10));
}
@@ -160,17 +156,17 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
Person person = new Person(address, "John Snow");
this.template.opsForStream()
.createGroup(STREAM_KEY, this.redisStreamMessageProducer.getBeanName())
.createGroup(STREAM_KEY, this.reactiveRedisStreamProducer.getBeanName())
.as(StepVerifier::create)
.assertNext(message -> assertThat(message).isEqualTo("OK"))
.thenCancel()
.verify(Duration.ofSeconds(10));
this.redisStreamMessageProducer.setCreateConsumerGroup(false);
this.redisStreamMessageProducer.setConsumerName(CONSUMER);
this.redisStreamMessageProducer.setReadOffset(ReadOffset.latest());
this.redisStreamMessageProducer.afterPropertiesSet();
this.redisStreamMessageProducer.start();
this.reactiveRedisStreamProducer.setCreateConsumerGroup(false);
this.reactiveRedisStreamProducer.setConsumerName(CONSUMER);
this.reactiveRedisStreamProducer.setReadOffset(ReadOffset.latest());
this.reactiveRedisStreamProducer.afterPropertiesSet();
this.reactiveRedisStreamProducer.start();
StepVerifier stepVerifier =
Flux.from(this.fluxMessageChannel)
@@ -196,13 +192,13 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
String consumerGroup = "testGroup";
String consumerName = "testConsumer";
this.redisStreamMessageProducer.setCreateConsumerGroup(true);
this.redisStreamMessageProducer.setAutoAck(false);
this.redisStreamMessageProducer.setConsumerGroup(consumerGroup);
this.redisStreamMessageProducer.setConsumerName(consumerName);
this.redisStreamMessageProducer.setReadOffset(ReadOffset.latest());
this.redisStreamMessageProducer.afterPropertiesSet();
this.redisStreamMessageProducer.start();
this.reactiveRedisStreamProducer.setCreateConsumerGroup(true);
this.reactiveRedisStreamProducer.setAutoAck(false);
this.reactiveRedisStreamProducer.setConsumerGroup(consumerGroup);
this.reactiveRedisStreamProducer.setConsumerName(consumerName);
this.reactiveRedisStreamProducer.setReadOffset(ReadOffset.latest());
this.reactiveRedisStreamProducer.afterPropertiesSet();
this.reactiveRedisStreamProducer.start();
AtomicReference<SimpleAcknowledgment> acknowledgmentReference = new AtomicReference<>();
@@ -239,6 +235,65 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
.verifyComplete();
}
@Autowired
ReactiveRedisStreamMessageProducer reactiveErrorRedisStreamProducer;
@Autowired
PollableChannel redisStreamErrorChannel;
@Test
@RedisAvailable
public void testReadingNextMessagesWhenSerializationException() {
Person person = new Person(new Address("Winterfell, Westeros"), "John Snow");
Date testDate = new Date();
this.reactiveErrorRedisStreamProducer.start();
StepVerifier stepVerifier =
Flux.from(this.fluxMessageChannel)
.map(Message::getPayload)
.cast(Date.class)
.as(StepVerifier::create)
.expectNext(testDate)
.thenCancel()
.verifyLater();
this.messageHandler.handleMessage(new GenericMessage<>(person));
Message<?> errorMessage = this.redisStreamErrorChannel.receive(10_000);
assertThat(errorMessage).isInstanceOf(ErrorMessage.class)
.extracting("payload.message")
.asInstanceOf(InstanceOfAssertFactories.STRING)
.contains("Cannot deserialize Redis Stream Record")
.contains("Cannot parse date out of");
Mono<PendingMessagesSummary> pendingMessage =
template.opsForStream()
.pending(STREAM_KEY, this.reactiveErrorRedisStreamProducer.getBeanName());
StepVerifier.create(pendingMessage)
.assertNext(pendingMessagesSummary ->
assertThat(pendingMessagesSummary.getTotalPendingMessages()).isEqualTo(1L))
.verifyComplete();
Message<?> failedMessage = ((MessagingException) errorMessage.getPayload()).getFailedMessage();
StaticMessageHeaderAccessor.getAcknowledgment(failedMessage).acknowledge();
pendingMessage =
template.opsForStream()
.pending(STREAM_KEY, this.reactiveErrorRedisStreamProducer.getBeanName());
StepVerifier.create(pendingMessage)
.assertNext(pendingMessagesSummary ->
assertThat(pendingMessagesSummary.getTotalPendingMessages()).isEqualTo(0))
.verifyComplete();
this.messageHandler.handleMessage(new GenericMessage<>(testDate));
stepVerifier.verify(Duration.ofSeconds(10));
this.reactiveErrorRedisStreamProducer.stop();
}
@Configuration
static class ContextConfig {
@@ -263,6 +318,30 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
return new FluxMessageChannel();
}
@Bean
PollableChannel redisStreamErrorChannel() {
return new QueueChannel();
}
@Bean
ReactiveRedisStreamMessageProducer reactiveErrorRedisStreamProducer() {
ReactiveRedisStreamMessageProducer messageProducer =
new ReactiveRedisStreamMessageProducer(RedisAvailableRule.connectionFactory, STREAM_KEY);
messageProducer.setStreamReceiverOptions(
StreamReceiver.StreamReceiverOptions.builder()
.pollTimeout(Duration.ofMillis(100))
.targetType(Date.class)
.build());
messageProducer.setCreateConsumerGroup(true);
messageProducer.setAutoAck(false);
messageProducer.setConsumerName("testConsumer");
messageProducer.setReadOffset(ReadOffset.latest());
messageProducer.setAutoStartup(false);
messageProducer.setOutputChannel(fluxMessageChannel());
messageProducer.setErrorChannel(redisStreamErrorChannel());
return messageProducer;
}
@Bean
ReactiveRedisStreamMessageProducer reactiveRedisStreamProducer() {
ReactiveRedisStreamMessageProducer messageProducer =