SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1459: Pollable Consumer and Requeue

Browse Source 
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
This commit is contained in:
Gary Russell
2018-09-06 12:48:40 -04:00
committed by Oleg Zhurakousky
parent e4a82107d8
commit d7e88a3a9c
5 changed files with 141 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc
View File

@@ -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 "`<<binder-error-channels>>`".
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 <<polled-errors>>.
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<Map<String, Foo>>() {});
----
[[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 (`<destination>.<group>.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

30
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java
View File

@@ -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<Object> 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 <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
if (DefaultPollableMessageSource.this.recoveryCallback != null) {

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java
View File

@@ -38,7 +38,9 @@ public interface PollableSource<H> {
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.

47
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/RequeueCurrentMessageException.java Normal file
View File

@@ -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);
}
}

52
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java
View File

@@ -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<Object> 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);