GH-1952: Pollable Source Error Handling

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1952

Requeue if the error flow throws `RequeueCurrentMessageException`.

Resolves #1953

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

@@ -218,21 +218,19 @@ public class DefaultPollableMessageSource
 		}
 		catch (MessagingException e) {
 			if (this.retryTemplate == null && !shouldRequeue(e)) {
-				this.messagingTemplate.send(this.errorChannel, this.errorMessageStrategy
-						.buildErrorMessage(e, attributesHolder.get()));
-				return true;
-			}
-			else if (!ackCallback.isAcknowledged() && shouldRequeue(e)) {
-				AckUtils.requeue(ackCallback);
+				try {
+					this.messagingTemplate.send(this.errorChannel, this.errorMessageStrategy
+							.buildErrorMessage(e, attributesHolder.get()));
+				}
+				catch (MessagingException e1) {
+					requeueOrNack(message, ackCallback, e1);
+				}
 				return true;
 			}
 			else {
-				AckUtils.autoNack(ackCallback);
+				requeueOrNack(message, ackCallback, e);
+				return true;
 			}
-			if (e.getFailedMessage().equals(message)) {
-				throw e;
-			}
-			throw new MessageHandlingException(message, e);
 		}
 		catch (Exception e) {
 			AckUtils.autoNack(ackCallback);
@@ -247,6 +245,19 @@ public class DefaultPollableMessageSource
 		}
 	}
 
+	private void requeueOrNack(Message<?> message, AcknowledgmentCallback ackCallback, 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);
+		}
+	}
+
 	protected boolean shouldRequeue(Exception e) {
 		boolean requeue = false;
 		Throwable t = e.getCause();

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

@@ -40,6 +40,7 @@ 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.channel.DirectChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.messaging.Message;
@@ -395,8 +396,7 @@ public class PollableConsumerTests {
 			}
 
 		});
-		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(
-				null);
+		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
 		properties.setMaxAttempts(2);
 		properties.setBackOffInitialInterval(0);
 		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
@@ -414,6 +414,47 @@ public class PollableConsumerTests {
 		verify(callback).acknowledge(Status.REQUEUE);
 	}
 
+	@Test
+	public void testRequeueFromErrorFlow() {
+		TestChannelBinder binder = createBinder();
+		MessageConverterConfigurer configurer = this.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(1);
+		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		SubscribableChannel errorChannel = new DirectChannel();
+		errorChannel.subscribe(msg -> {
+			throw new RequeueCurrentMessageException((Throwable) msg.getPayload());
+		});
+		pollableSource.setErrorChannel(errorChannel);
+		try {
+			pollableSource.poll(received -> {
+				throw new RuntimeException("test requeue from error flow");
+			});
+		}
+		catch (Exception e) {
+			// no op
+		}
+		verify(callback).acknowledge(Status.REQUEUE);
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testAutoStartupOff() {