From b1a383060d0940890816f9a50155e6bce972e53a Mon Sep 17 00:00:00 2001 From: rohan mukesh Date: Tue, 13 Oct 2020 13:33:17 -0500 Subject: [PATCH] Add Error Handling to RedisStreamMessageProducer * 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 --- .../endpoint/MessageProducerSupport.java | 6 +- .../ReactiveRedisStreamMessageProducer.java | 51 +++--- ...activeRedisStreamMessageProducerTests.java | 151 +++++++++++++----- 3 files changed, 151 insertions(+), 57 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index e88f4bd0b6..8182a89afe 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -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); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java index 6c27700650..e19805fa39 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java @@ -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> 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) 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 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(); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java index 3bffa4d987..0909eacb79 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java @@ -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 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 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 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 =