From d7e88a3a9cf367a9152ccef2b4eeb428b1eec42f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 6 Sep 2018 12:48:40 -0400 Subject: [PATCH] GH-1459: Pollable Consumer and Requeue Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1459 Add a mechanism to cause a requeue of the current message. Polishing - PR Comments - simpler logic, no configuration needed. Polishing Resolves #1467 --- .../spring-cloud-stream-overview.adoc | 17 +++++- .../binder/DefaultPollableMessageSource.java | 30 ++++++++++- .../cloud/stream/binder/PollableSource.java | 4 +- .../RequeueCurrentMessageException.java | 47 +++++++++++++++++ .../stream/binder/PollableConsumerTests.java | 52 +++++++++++++++++-- 5 files changed, 141 insertions(+), 9 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/RequeueCurrentMessageException.java diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 91fa54261..0e8376509 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -664,6 +664,8 @@ To do that Spring Cloud Function allows you to use `|` (pipe) symbol. So to fini [[spring-cloud-streams-overview-using-polled-consumers]] ==== Using Polled Consumers +===== Overview + When using polled consumers, you poll the `PollableMessageSource` on demand. Consider the following example of a polled consumer: @@ -697,7 +699,7 @@ public ApplicationRunner poller(PollableMessageSource destIn, MessageChannel des } } catch (Exception e) { - // handle failure (throw an exception to reject the message); + // handle failure } } }; @@ -710,7 +712,7 @@ It returns `true` if the message was received and successfully processed. As with message-driven consumers, if the `MessageHandler` throws an exception, messages are published to error channels, as discussed in "`<>`". Normally, the `poll()` method acknowledges the message when the `MessageHandler` exits. -If the method exits abnormally, the message is rejected (not re-queued). +If the method exits abnormally, the message is rejected (not re-queued), but see <>. You can override that behavior by taking responsibility for the acknowledgment, as shown in the following example: [source,java] @@ -754,6 +756,17 @@ boolean result = pollableSource.poll(received -> { }, new ParameterizedTypeReference>() {}); ---- +[[polled-errors]] +===== Handling Errors + +By default, an error channel is configured for the pollable source; if the callback throws an exception, an `ErrorMessage` is sent to the error channel (`..errors`); this error channel is also bridged to the global Spring Integration `errorChannel`. + +You can subscribe to either error channel with a `@ServiceActivator` to handle errors; without a subscription, the error will simply be logged and the message will be acknowledged as successful. +If the error channel service activator throws an exception, the message will be rejected (by default) and won't be redelivered. +If the service activator throws a `RequeueCurrentMessageException`, the message will be requeued at the broker and will be again retrieved on a subsequent poll. + +If the listener throws a `RequeueCurrentMessageException` directly, the message will be requeued, as discussed above, and will not be sent to the error channels. + [[spring-cloud-stream-overview-error-handling]] === Error Handling diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index 727522dd1..6db615529 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -42,6 +42,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.support.ChannelInterceptor; @@ -135,7 +136,12 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life } public void setRecoveryCallback(RecoveryCallback recoveryCallback) { - this.recoveryCallback = recoveryCallback; + this.recoveryCallback = context -> { + if (!shouldRequeue((MessagingException) context.getLastThrowable())) { + return recoveryCallback.recover(context); + } + throw (MessagingException) context.getLastThrowable(); + }; } public void setErrorChannel(MessageChannel errorChannel) { @@ -217,6 +223,18 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life } return true; } + catch (MessagingException e) { + if (!ackCallback.isAcknowledged() && shouldRequeue(e)) { + AckUtils.requeue(ackCallback); + } + else { + AckUtils.autoNack(ackCallback); + } + if (e.getFailedMessage().equals(message)) { + throw e; + } + throw new MessageHandlingException(message, e); + } catch (Exception e) { AckUtils.autoNack(ackCallback); if (e instanceof MessageHandlingException @@ -230,6 +248,16 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life } } + protected boolean shouldRequeue(Exception e) { + boolean requeue = false; + Throwable t = e.getCause(); + while (t != null && !requeue) { + requeue = t instanceof RequeueCurrentMessageException; + t = t.getCause(); + } + return requeue; + } + @Override public boolean open(RetryContext context, RetryCallback callback) { if (DefaultPollableMessageSource.this.recoveryCallback != null) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java index 9faf8149f..b9ecc0f16 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java @@ -38,7 +38,9 @@ public interface PollableSource { boolean poll(H handler); /** - * Poll the consumer and convert the payload to the type. + * Poll the consumer and convert the payload to the type. Throw a + * {@code RequeueCurrentMessageException} to force the current message to be requeued + * in the broker (after retries are exhausted, if configured). * @param handler the handler. * @param type the type. * @return true if a message was handled. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/RequeueCurrentMessageException.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/RequeueCurrentMessageException.java new file mode 100644 index 000000000..8b58fc5a9 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/RequeueCurrentMessageException.java @@ -0,0 +1,47 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +/** + * When using a {@code PollableMessageSource} throw this exception to cause the current + * message to be requeued in the broker so that it will be redelivered on the next poll. + * + * @author Gary Russell + * @since 2.1 + * + */ +public class RequeueCurrentMessageException extends RuntimeException { + + private static final long serialVersionUID = 1L; + + public RequeueCurrentMessageException() { + super(); + } + + public RequeueCurrentMessageException(String message, Throwable cause) { + super(message, cause); + } + + public RequeueCurrentMessageException(String message) { + super(message); + } + + public RequeueCurrentMessageException(Throwable cause) { + super(cause); + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java index 665082f91..42ffa3aa8 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java @@ -36,19 +36,24 @@ import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.ApplicationContext; import org.springframework.core.ParameterizedTypeReference; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.acks.AcknowledgmentCallback; +import org.springframework.integration.acks.AcknowledgmentCallback.Status; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.converter.SmartMessageConverter; -import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.MimeType; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; /** * @author Gary Russell @@ -74,7 +79,7 @@ public class PollableConsumerTests { DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); - pollableSource.addInterceptor(new ChannelInterceptorAdapter() { + pollableSource.addInterceptor(new ChannelInterceptor() { @Override public Message preSend(Message message, MessageChannel channel) { @@ -229,7 +234,7 @@ public class PollableConsumerTests { properties.setHeaderMode(HeaderMode.embeddedHeaders); DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); - pollableSource.addInterceptor(new ChannelInterceptorAdapter() { + pollableSource.addInterceptor(new ChannelInterceptor() { @Override public Message preSend(Message message, MessageChannel channel) { @@ -253,7 +258,7 @@ public class PollableConsumerTests { DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); - pollableSource.addInterceptor(new ChannelInterceptorAdapter() { + pollableSource.addInterceptor(new ChannelInterceptor() { @Override public Message preSend(Message message, MessageChannel channel) { @@ -289,7 +294,7 @@ public class PollableConsumerTests { DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); - pollableSource.addInterceptor(new ChannelInterceptorAdapter() { + pollableSource.addInterceptor(new ChannelInterceptor() { @Override public Message preSend(Message message, MessageChannel channel) { @@ -314,6 +319,43 @@ public class PollableConsumerTests { assertThat(count.get()).isEqualTo(1); } + @Test + public void testRequeue() { + TestChannelBinder binder = createBinder(); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); + AcknowledgmentCallback callback = mock(AcknowledgmentCallback.class); + pollableSource.addInterceptor(new ChannelInterceptor() { + + @Override + public Message preSend(Message message, MessageChannel channel) { + return MessageBuilder.fromMessage(message) + .setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, callback) + .build(); + } + + }); + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); + properties.setMaxAttempts(2); + properties.setBackOffInitialInterval(0); + binder.bindPollableConsumer("foo", "bar", pollableSource, properties); + final AtomicInteger count = new AtomicInteger(); + try { + assertThat(pollableSource.poll(received -> { + count.incrementAndGet(); + throw new RequeueCurrentMessageException("test retry"); + })).isTrue(); + fail("Expected exception"); + } + catch (Exception e) { + // no op + } + assertThat(count.get()).isEqualTo(2); + verify(callback).acknowledge(Status.REQUEUE); + } + private TestChannelBinder createBinder(String... args) { this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration.getCompleteConfiguration()) .web(WebApplicationType.NONE).run(args);