About Pascal Alma

Pascal is a senior JEE Developer and Architect at 4Synergy in The Netherlands. Pascal has been designing and building J2EE applications since 2001. He is particularly interested in Open Source toolstack (Mule, Spring Framework, JBoss) and technologies like Web Services, SOA and Cloud technologies. Specialties: JEE, SOA, Mule ESB, Maven, Cloud Technology, Amazon AWS.

Mule ESB, ActiveMQ and the DLQ

apache-activemq-logo
In this post I will show a simple Mule ESB flow to see the DLQ feature of Active MQ in action.
I assume you have a running Apache ActiveMQ instance available (if not you can download a version here). In this example I make use of Mule ESB 3.4.2 and ActiveMQ 5.9.0. We can create a simple Mule project based on the following pom file:
 
 
 

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

    <modelVersion>4.0.0</modelVersion>
    <groupId>net.pascalalma.demo</groupId>
    <artifactId>activemq-test-flow</artifactId>
    <packaging>mule</packaging>
    <name>${project.artifactId}</name>
    <version>1.0.0-SNAPSHOT</version>
    <properties>
        <mule.version>3.4.2</mule.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <jdk.version>1.7</jdk.version>
        <junit.version>4.9</junit.version>
        <activemq.version>5.9.0</activemq.version>
    </properties>
    <dependencies>
        <!-- Mule Dependencies -->
        <dependency>
            <groupId>org.mule</groupId>
            <artifactId>mule-core</artifactId>
            <version>${mule.version}</version>
        </dependency>
        <!-- Mule Transports -->
        <dependency>
            <groupId>org.mule.transports</groupId>
            <artifactId>mule-transport-jms</artifactId>
            <version>${mule.version}</version>
        </dependency>
        <dependency>
            <groupId>org.mule.transports</groupId>
            <artifactId>mule-transport-vm</artifactId>
            <version>${mule.version}</version>
        </dependency>
        <!-- Mule Modules -->
        <dependency>
            <groupId>org.mule.modules</groupId>
            <artifactId>mule-module-client</artifactId>
            <version>${mule.version}</version>
        </dependency>
        <dependency>
            <groupId>org.mule.modules</groupId>
            <artifactId>mule-module-scripting</artifactId>
            <version>${mule.version}</version>
        </dependency>
        <!-- for testing -->
        <dependency>
            <groupId>org.mule.tests</groupId>
            <artifactId>mule-tests-functional</artifactId>
            <version>${mule.version}</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>${junit.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-client</artifactId>
            <version>${activemq.version}</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.3.2</version>
                <configuration>
                    <source>${jdk.version}</source>
                    <target>${jdk.version}</target>
                    <encoding>${project.build.sourceEncoding}</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-resources-plugin</artifactId>
                <version>2.5</version>
                <configuration>
                    <encoding>${project.build.sourceEncoding}</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.mule.tools</groupId>
                <artifactId>maven-mule-plugin</artifactId>
                <version>1.9</version>
                <extensions>true</extensions>
                <configuration>
                    <copyToAppsDirectory>false</copyToAppsDirectory>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

There is not much special here. Besides the necessary dependencies I have added the maven-mule-plugin so I can create a ‘mule’ packaging type and run Mule from my IDE.
With this Maven pom in place we can create the following two Mule configurations. One for the Mule flow to test our transaction:

<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns="http://www.mulesoft.org/schema/mule/core"
      xmlns:scripting="http://www.mulesoft.org/schema/mule/scripting"
      version="EE-3.4.1"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="
		http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
		http://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/current/mule-scripting.xsd">
    <flow name="MainFlow">
        <inbound-endpoint ref="event-queue" />
        <logger category="net.pascalalma.demo.MainFlow" level="INFO" message="Received message from activeMQ" />
        <scripting:component>
            <scripting:script engine="Groovy">
                throw new Exception('Soap Fault Response detected')
            </scripting:script>
        </scripting:component>
        <outbound-endpoint ref="result-queue" />
    </flow>
</mule>

In this flow we receive a message from the inbound endpoint, log a message and throw an exception before the message is put on the next queue. As we can see I didn’t add any exception handler. The configuration of the endpoints and connectors look like this:

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns="http://www.mulesoft.org/schema/mule/core"
      xmlns:jms="http://www.mulesoft.org/schema/mule/jms"
      xmlns:spring="http://www.springframework.org/schema/beans"
      version="EE-3.4.1"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="
		http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
		http://www.mulesoft.org/schema/mule/jms http://www.mulesoft.org/schema/mule/jms/current/mule-jms.xsd">

    <spring:bean id="redeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
        <spring:property name="maximumRedeliveries" value="5"/>
        <spring:property name="initialRedeliveryDelay" value="500"/>
        <spring:property name="maximumRedeliveryDelay" value="10000"/>
        <spring:property name="useExponentialBackOff" value="false"/>
        <spring:property name="backOffMultiplier" value="3"/>
    </spring:bean>
    <!-- ActiveMQ Connection factory -->
    <spring:bean id="amqFactory" class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true">
        <spring:property name="brokerURL" value="tcp://localhost:61616" />
        <spring:property name="redeliveryPolicy" ref="redeliveryPolicy" />
    </spring:bean>
    <jms:activemq-connector name="activeMqConnector"
                            connectionFactory-ref="amqFactory"
                            persistentDelivery="true"
                            numberOfConcurrentTransactedReceivers="2"
                            specification="1.1" />
    <jms:endpoint name="event-queue" connector-ref="activeMqConnector" queue="event-queue" >
        <jms:transaction action="ALWAYS_BEGIN" />
    </jms:endpoint>
    <jms:endpoint name="result-queue" connector-ref="activeMqConnector" queue="result-queue" >
        <jms:transaction action="ALWAYS_JOIN" />
    </jms:endpoint>
</mule>

I defined a Spring bean for an ActiveMQ connection factory and one for the redelivery policy of this factory. With this redelivery policy we can configure how often Mule should retry to process a message from the queue when the original attempt failed. A nice feature in the redelivery policy is the ‘backOffMultiplier’ and ‘useExponentialBackOff’ combination. With these options you can have the period between two redelivery attempts increase exponentially until ‘maximumRedeliveryDelay’ is reached. In that case Mule will wait the ‘maximumRedeliveryDelay’ for the next attempt.

So with these configurations we can create a Mule test class and run it. The test class would look something like this:

package net.pascalalma.demo;

import org.junit.Test;
import org.mule.DefaultMuleMessage;
import org.mule.api.MuleMessage;
import org.mule.module.client.MuleClient;
import org.mule.tck.junit4.FunctionalTestCase;

public class TransactionFlowTest extends FunctionalTestCase {

    @Override
    protected String getConfigResources() {
        return "app/test-flow.xml, app/test-endpoints.xml";
    }

    @Test
    public void testError() throws Exception {
        MuleClient client = new MuleClient(muleContext);
        MuleMessage inMsg = new DefaultMuleMessage("<txt>Some message</txt>", muleContext);
        client.dispatch("event-queue", inMsg);

        // Give Mule the chance to redeliver the message
        Thread.sleep(4000);
    }
}

If we run this test you will see messages in the logging like:

Exception stack is:
1. "Message with id "ID:Pascals-MacBook-Pro-2.local-59158-1406440948059-1:1:3:1:1" has been redelivered 3 times on endpoint "jms://event-queue", which exceeds the maxRedelivery setting of 0 on the connector "activeMqConnector". Message payload is of type: ActiveMQTextMessage (org.mule.transport.jms.redelivery.MessageRedeliveredException)
  org.mule.transport.jms.redelivery.JmsXRedeliveryHandler:87 (http://www.mulesoft.org/docs/site/current3/apidocs/org/mule/transport/jms/redelivery/MessageRedeliveredException.html)

If we now switch to the ActiveMQ console which can be reached at http://localhost:8161 for the default local installation we can see the following queues:
 
screenshot-2014-07-27-10-08-09
As expected we see two queues being created, the event-queue which is empty and the default ActiveMQ.DLQ which contains our message:
 
screenshot-2014-07-27-10-13-54

As you can image it might be handy to have a specific DLQ for each queue instead of one DLQ which will contain all kinds of undeliverable messages. Luckily this is easy to configure in ActiveMQ. Just put the following in the ‘activemq.xml’ file that can be found in ‘$ACTIVEMQ_HOME/conf’ folder.

<!-- Set the following policy on all queues using the '>' wildcard -->
<policyEntry queue=">">
  <deadLetterStrategy>
    <individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true" />
  </deadLetterStrategy>
</policyEntry>

If we now restart ActiveMQ, remove the existing queues and rerun our test we see the following result:
 
screenshot-2014-07-27-10-29-22
So with this setup each queue has its own DLQ. For more options regarding these ActieMQ settings see here. With the Mule flow created in this post it is easy to test and play with these settings.

Reference: Mule ESB, ActiveMQ and the DLQ from our JCG partner Pascal Alma at the The Pragmatic Integrator blog.

Do you want to know how to develop your skillset to become a Java Rockstar?

Subscribe to our newsletter to start Rocking right now!

To get you started we give you two of our best selling eBooks for FREE!

JPA Mini Book

Learn how to leverage the power of JPA in order to create robust and flexible Java applications. With this Mini Book, you will get introduced to JPA and smoothly transition to more advanced concepts.

JVM Troubleshooting Guide

The Java virtual machine is really the foundation of any Java EE platform. Learn how to master it with this advanced guide!

Given email address is already subscribed, thank you!
Oops. Something went wrong. Please try again later.
Please provide a valid email address.
Thank you, your sign-up request was successful! Please check your e-mail inbox.
Please complete the CAPTCHA.
Please fill in the required fields.

Leave a Reply


7 − = five



Java Code Geeks and all content copyright © 2010-2014, Exelixis Media Ltd | Terms of Use | Privacy Policy | Contact
All trademarks and registered trademarks appearing on Java Code Geeks are the property of their respective owners.
Java is a trademark or registered trademark of Oracle Corporation in the United States and other countries.
Java Code Geeks is not connected to Oracle Corporation and is not sponsored by Oracle Corporation.
Do you want to know how to develop your skillset and become a ...
Java Rockstar?

Subscribe to our newsletter to start Rocking right now!

To get you started we give you two of our best selling eBooks for FREE!

Get ready to Rock!
You can download the complementary eBooks using the links below:
Close