I am working on spring-boot rabbitMQ. I am creating a dead letter queue which i can able to see in RabbitMQ admin as "D,DLE" but no DLK might be i am missing setting "x-dead-letter-routing-key", the thing is i don't want routing key.Few of my consumers are binding to a particular exchange and i am creating DLE per exchange if there is any issue in a consumer for that exchange then DLE attached to that exchange which receive that message and do a user dependent logic.But unfortunately this is not working, the DLE isn't receiving any message.
Please find the code below,
package com.sample.rabbit;
import org.slf4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Argument;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.converter.DefaultClassMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.util.ErrorHandler;
@SpringBootApplication
public class SampleRabbitApplication {
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(SampleRabbitApplication.class, args);
context.getBean(SampleRabbitApplication.class).runDemo();
context.close();
}
@Autowired
private RabbitTemplate template;
private void runDemo() throws Exception {
this.template.convertAndSend("sample-queue", new Foo("bar"),m -> {
m.getMessageProperties().setHeader("__TypeId__","foo");
return m;
});
this.template.convertAndSend("sample-queue", new Foo("throw"),m -> {
m.getMessageProperties().setHeader("__TypeId__","foo");
return m;
});
this.template.convertAndSend("sample-queue", new Foo("bar"), m -> {
return new Message("some bad json".getBytes(), m.getMessageProperties());
});
Thread.sleep(5000);
}
@RabbitListener(
id = "sample-queue",
bindings = @QueueBinding(
value = @org.springframework.amqp.rabbit.annotation.Queue(value = "sample-queue", durable = "true"),
exchange = @org.springframework.amqp.rabbit.annotation.Exchange(value = "sample.exchange", durable = "true")
)
)
public void handle(Foo in) {
System.out.println("Received: " + in);
if("throw".equalsIgnoreCase(in.getFoo())){
throw new BadRequestException("Foo contains throw so it throwed the exception.");
}
}
@RabbitListener(
id = "sample-dead-letter-queue",
bindings = @QueueBinding(
value = @org.springframework.amqp.rabbit.annotation.Queue(value = "sample-dead-letter-queue", durable = "true", arguments = {@Argument(name = "x-dead-letter-exchange",value = "sample.exchange"),@Argument(name = "x-dead-letter-routing-key",value = "#")}),
exchange = @org.springframework.amqp.rabbit.annotation.Exchange(value = "critical.exchange", durable = "true",type = "topic")
)
)
public void handleDLE(Message in) {
System.out.println("Received in DLE: " + in.getBody());
}
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(jsonConverter());
factory.setErrorHandler(errorHandler());
return factory;
}
@Bean
public ErrorHandler errorHandler() {
return new ConditionalRejectingErrorHandler(new MyFatalExceptionStrategy());
}
@Bean
public MessageConverter jsonConverter() {
Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
DefaultClassMapper mapper = new DefaultClassMapper();
mapper.setDefaultType(Foo.class);
converter.setClassMapper(mapper);
return new Jackson2JsonMessageConverter();
}
public static class MyFatalExceptionStrategy extends ConditionalRejectingErrorHandler.DefaultExceptionStrategy {
private final Logger LOG = org.slf4j.LoggerFactory.getLogger(getClass());
public boolean isFatal(Throwable t) {
if (t instanceof ListenerExecutionFailedException && isCauseFatal(t.getCause())) {
//To do : Here we have to configure DLE(Critical queue) and put all the messages in the critical queue.
ListenerExecutionFailedException lefe = (ListenerExecutionFailedException) t;
if(lefe.getFailedMessage() != null) {
LOG.info("Failed to process inbound message from queue "
+ lefe.getFailedMessage().getMessageProperties().getConsumerQueue()
+ "; failed message: " + lefe.getFailedMessage(), t);
} else {
LOG.info("Failed to process inbound message from queue "
+ lefe.getMessage(), t);
}
}
return super.isFatal(t);
}
private boolean isCauseFatal(Throwable cause) {
return cause instanceof MessageConversionException
|| cause instanceof org.springframework.messaging.converter.MessageConversionException
|| cause instanceof MethodArgumentNotValidException
|| cause instanceof MethodArgumentTypeMismatchException
|| cause instanceof NoSuchMethodException
|| cause instanceof ClassCastException
|| isUserCauseFatal(cause);
}
/**
* Subclasses can override this to add custom exceptions.
* @param cause the cause
* @return true if the cause is fatal.
*/
protected boolean isUserCauseFatal(Throwable cause) {
return true;
}
}
public static class Foo {
private String foo;
public Foo() {
super();
}
public Foo(String foo) {
this.foo = foo;
}
public String getFoo() {
return this.foo;
}
public void setFoo(String foo) {
this.foo = foo;
}
@Override
public String toString() {
return "Foo [foo=" + this.foo + "]";
}
}
}
My Exchanges and Queue's are Direct,each of my consumer will use different routing key but it belongs to same exchange, So how can i write a DLE which efficiently consume all the failure messages.In the above code sample one message is success and other one failure but i couldn't see the failure message in DLE.
Any help would be appreciated.
If you configure a queue with a dead-letter exchange (DLX) but no dead-letter routing key, the message is routed to the DLX with the original routing key.
The easiest way to handle your use case is to make the DLX a topic exchange and bind a queue to it with routing key # (wildcard for all messages) and all errors will go to that queue.
If you want to segregate the errors into individual queues, then bind a DLQ for each with the original routing key.
EDIT
Here is the correct configuration:
@RabbitListener(id = "sample-queue",
bindings = @QueueBinding(value = @Queue(value = "sample-queue", durable = "true", arguments =
@Argument(name = "x-dead-letter-exchange", value = "critical.exchange")),
exchange = @Exchange(value = "sample.exchange", durable = "true")))
public void handle(Foo in) {
System.out.println("Received: " + in);
}
@RabbitListener(id = "sample-dead-letter-queue", containerFactory = "noJsonContainerFactory",
bindings = @QueueBinding(value = @Queue(value = "sample-dead-letter-queue", durable = "true"),
exchange = @Exchange(value = "critical.exchange", durable = "true", type = "topic"),
key = "#"))
public void handleDLE(Message in) {
System.out.println("Received in DLE: " + new String(in.getBody()));
}
@Bean
public SimpleRabbitListenerContainerFactory noJsonContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setErrorHandler(errorHandler());
return factory;
}