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

GH-1578 Fixed error handling in DefaultPollableMessageSource

Browse Source 
ensured that RequeueCurrentMessageException is treated differently where message does not end up in error channel
Resolves #1578
This commit is contained in:
Oleg Zhurakousky
2019-01-21 15:30:27 +01:00
parent 794c75f536
commit ea594d1ebd
3 changed files with 11 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-stream-reactive/pom.xml
View File

@@ -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>

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

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

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

@@ -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