ActiveMQ - Redelivery policy and Dead letter queue for topic

Question

I'm working with ActiveMQ Artemis 2.17 and Spring Boot 2.5.7. I'm publishing messages on topic and queue and consuming it. All is done through JMS. All queues (anycast or multicast) are durables. My topic (multicast address) has 2 durable queues in order to have 2 separate consumers. For my topic, the 2 consumers use durable and shared subscriptions (JMS 2.0). All processing is transactional, managed through Atomikos transaction manager (I need it for a 2 phases commit with the database).

I have a problem with the redelivery policy and DLQ. When I throw an exception during the processing of the message, the redelivery policy applies correctly for the queue (anycast queue) and a DLQ is created with the message. However, for the topic (multicast queue), the redelivery policy does not apply and the message is not sent into a DLQ.

Here is my ActiveMQ Artemis broker configuration:




    

        0.0.0.0


        
        
        org.apache.activemq.artemis.utils.DefaultSensitiveStringCodec;key=UBNTd0dS9w6f8HDIyGW9


        
        true
        
            
                BINDINGS_TABLE
                MESSAGE_TABLE
                PAGE_TABLE
                LARGE_MESSAGES_TABLE
                NODE_MANAGER_TABLE
                2000
                20000
                5
                
                
                    
                    
                    
                    
                    
                    
                    
                
            
        


        
        
        
            
                DLA
                true
                
                .DLQ
                ExpiryQueue
                false
                
                .EXP
                -1
                5
                250
                2.0
                0.5
                10000
                100000
                20000
                5
                -1
                PAGE
                0
                false
                false
                false
                0
                -1
                0
                true
                -1
                NOTIFY
                5
                
                false
                true
                false
                30000
                0
                OFF
                
                false
                true
                30000
                OFF
                200
                false
                -1
                ANYCAST
                ANYCAST
                -1
                0
                true
                
                true
                true
                1024
            
        


        
        
        
            
            
            
                tcp://0.0.0.0:61616?protocols=CORE,tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;useEpoll=true,cacheDestinations=true
            
        


        
        
        
            tcp://localhost:61616
            tcp://localhost:61616
            tcp://localhost:61617
        

        
        
            
                
                    true
                
            
        

        
            
                netty-connector
                
                    broker1-connector
                    broker2-connector
                
            
        

        
        

        
        true
        120000
        60000
        HALT


        72000


        

        
        false

        
        
        
            
                
                    
                        true
                    
                    
                        true
                    
                
            
            
                
                    
                        true

The JMS configuration into the Spring Boot is the following:

@Bean
public DynamicDestinationResolver destinationResolver() {
    return new DynamicDestinationResolver() {
        @Override
        public Destination resolveDestinationName(Session session, String destinationName, boolean pubSubDomain) throws JMSException {
            pubSubDomain = destinationName.startsWith("topic.");            
            return super.resolveDestinationName(session, destinationName, pubSubDomain);
        }
    };
}

@Bean
public JmsListenerContainerFactory queueConnectionFactory(ConnectionFactory connectionFactory,
                                                             DefaultJmsListenerContainerFactoryConfigurer configurer,
                                                             JmsErrorHandler jmsErrorHandler) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    configurer.configure(factory, connectionFactory);
    factory.setPubSubDomain(false);
    factory.setSessionTransacted(true);
    factory.setErrorHandler(jmsErrorHandler);

    return factory;
}

@Bean
public JmsListenerContainerFactory topicConnectionFactory(ConnectionFactory connectionFactory,
                                                             DefaultJmsListenerContainerFactoryConfigurer configurer,
                                                             JmsErrorHandler jmsErrorHandler) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    // This provides all boot's default to this factory, including the message converter
    configurer.configure(factory, connectionFactory);
    factory.setPubSubDomain(true);
    factory.setSessionTransacted(true);
    factory.setSubscriptionDurable(true);
    factory.setSubscriptionShared(true);
    factory.setErrorHandler(jmsErrorHandler);

    return factory;
}

The messages publisher:

@GetMapping("api/send")
public void sendDataToJms() throws InterruptedException {
    OrderRest currentService = this.applicationContext.getBean(OrderRest.class);
    for (Long i = 0L; i < 1L; i++) {
        currentService.sendTopicMessage(i);
        currentService.sendQueueMessage(i);
        Thread.sleep(200L);
    }
}

@Transactional
public void sendTopicMessage(Long id) {
    Order myMessage = new Order("--" + id.toString() + "--", new Date());
    jmsTemplate.convertAndSend("topic.test_rde", myMessage);
}

@Transactional
public void sendQueueMessage(Long id) {
    Order myMessage = new Order("--" + id.toString() + "--", new Date());
    jmsTemplate.convertAndSend("queue.test_rde", myMessage);
}

And the listeners:

@Transactional
@JmsListener(destination = "topic.test_rde", containerFactory = "topicConnectionFactory", subscription = "rde_receiver_1")
public void receiveMessage_rde_1(@Payload Order order, @Headers MessageHeaders headers, Message message, Session session) {
    log.info("---> Consumer1 - rde_receiver_1 - " + order.getContent());
    throw new ValidationException("Voluntary exception", "entity", List.of(), List.of());
}

@Transactional
@JmsListener(destination = "queue.test_rde", containerFactory = "queueConnectionFactory")
public void receiveMessage_rde_queue(@Payload Order order, @Headers MessageHeaders headers, Message message, Session session) {
    log.info("---> Consumer1 - rde_receiver_queue - " + order.getContent());
    throw new ValidationException("Voluntary exception", "entity", List.of(), List.of());
}

Is it normal that the redelivery policy and the DLQ mechanismd do only apply for a queue (anycat queue)?
Is it possible to also apply it for topic (multicast queue) and shared durable subscriptions?
If not, how can I have a topic behaviour but with redelivery and DLQ mechanisms? Should I use the divert solution of ActiveMQ?

ActiveMQ - Redelivery policy and Dead letter queue for topic

Answers (1)

Related Questions