spring-cloud / spring-cloud/spring-cloud-stream
ImmediateAcknowledgeAmqpException lost due to globalErrorChannelCustomizer bean.
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 1.1k
- Forks
- 646
- Avg merge
- 2d 3h
- Merged PRs (30d)
- 8
Description
Describe the issue
When handling a message from errorChannel and throwing ImmediateAcknowledgeAmqpException so that event is not placed in dlq, this exception is being lost because the following bean configuration:
@Bean
public static BeanPostProcessor globalErrorChannelCustomizer() {
return new BeanPostProcessor() {
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if ("errorChannel".equals(beanName)) {
((PublishSubscribeChannel) bean).setIgnoreFailures(true);
}
return bean;
}
};
}
is forcing to ignore failures. Hence ImmediateAcknowledgeAmqpException is not being propagated correctly and message remains in Rabbit dlq.
It was mentioned as a temporary solution: https://github.com/spring-cloud/spring-cloud-stream/issues/2066#issuecomment-766820249 to fetch the bean from applicationContext and override the bean's property. This temporary solution worked.
Suggestion
However, is it possible to improve the temporary solution by having @ConditionalOnProperty annotation on the bean for example?
- ServiceActivator
@ServiceActivator(inputChannel = "errorChannel")
public void error(MessagingException exception) {
if (removeMessageFromDlq(exception)) {
throw new ImmediateAcknowledgeAmqpException(exception);
}
log.info("Keeping message in dlq: {}", exception.toString());
}
- spring-integration method handling exception in BroadcastingDispatcher class:
private boolean invokeHandler(MessageHandler handler, Message<?> message) {
try {
handler.handleMessage(message);
return true;
} catch (RuntimeException var4) {
if (!this.ignoreFailures) {
if (var4 instanceof MessagingException && ((MessagingException)var4).getFailedMessage() == null) {
throw new MessagingException(message, "Failed to handle Message", var4);
} else {
throw var4;
}
} else {
if (this.logger.isWarnEnabled()) {
this.logger.warn("Suppressing Exception since 'ignoreFailures' is set to TRUE.", var4);
}
return false;
}
}
}
Please note I did try setting spring.integration.channels.error.ignoreFailures: false which didn't make a difference.
Version of the framework
spring-cloud-stream:3.2.4
spring-cloud-stream-binder-rabbit:3.2.4
spring-integration-core:5.5.16
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the globalErrorChannelCustomizer BeanPostProcessor and BroadcastingDispatcher.invokeHandler shown in the issue, then trace how errorChannel and ImmediateAcknowledgeAmqpException interact. Check why the spring.integration.channels.error.ignoreFailures setting has no effect. Done means the exception propagates correctly without the applicationContext override and the Rabbit DLQ behavior is covered by a regression test.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, rabbitmq, spring
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100