diff --git a/spring-cloud-stream-reactive/pom.xml b/spring-cloud-stream-reactive/pom.xml
index afdd00844..6be030a1a 100644
--- a/spring-cloud-stream-reactive/pom.xml
+++ b/spring-cloud-stream-reactive/pom.xml
@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-parent</artifactId>
-		<version>2.1.1.BUILD-SNAPSHOT</version>
+		<version>2.2.0.BUILD-SNAPSHOT</version>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>
 
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 6db615529..abb2b0a42 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -202,18 +202,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 				.getAcknowledgmentCallback(message);
 		try {
 			if (this.retryTemplate == null) {
-				if (this.errorChannel == null) {
-					this.handle(message, handler);
-				}
-				else {
-					try {
-						this.handle(message, handler);
-					}
-					catch (Exception e) {
-						this.messagingTemplate
-							.send(this.errorChannel, this.errorMessageStrategy.buildErrorMessage(e, attributesHolder.get()));
-					}
-				}
+				this.handle(message, handler);
 			}
 			else {
 				this.retryTemplate.execute(context -> {
@@ -224,8 +213,14 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 			return true;
 		}
 		catch (MessagingException e) {
-			if (!ackCallback.isAcknowledged() && shouldRequeue(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);
+				return true;
 			}
 			else {
 				AckUtils.autoNack(ackCallback);
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 a6882bd98..edc1bc0e0 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
@@ -356,7 +356,7 @@ public class PollableConsumerTests {
 				count.incrementAndGet();
 				throw new RequeueCurrentMessageException("test retry");
 			})).isTrue();
-			fail("Expected exception");
+			//fail("Expected exception"); // see GH-1578
 		}
 		catch (Exception e) {
 			// no op