Implement simple persistent redelivery with backoff mixing Apache Camel & ActiveMQ

When you use Apache Camel routes for your integration, when a failure occurs, a classic pattern is to retry the message. That’s especially true for the recoverable errors: for instance, if you have a network outage, you can replay the messages, hoping we will have a network recovery.

In Apache Camel, this redelivery policy is configured in the error handler. The default and dead letter error handlers supports such policy.
However, by default, the redelivered exchange is stored in memory, and new exchanges are not coming through since the first redelivered exchange with exception is not flagged as “handled”.

This approach could be an issue as if you restart the container hosting the Camel route (like Apache Karaf), the exchange is lost. More other, in term of performance, we might want to still get the exchanges going through.

There are several solutions to achieve this. In this blog, I will illustrate a possible implementation of a persistent redelivery policy with backoff support.

Apache Camel route

Let’s start with routes to illustrate the use case. I’m using the Camel Blueprint DSL in this example as my preferred container is Apache Karaf (of course 😉 ).

The first route is pretty simple: it’s listening for incoming HTTP request (using the Camel Jetty component), convert this as a String, and send to a JMS queue.

The second route consumes messages from the JMS queue, prepare a JSON using a processor (registered as a service in Karaf) and call a REST service.

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"           xmlns:cxf="http://camel.apache.org/schema/cxf">  <reference id="connectionFactory" interface="javax.jms.ConnectionFactory"/>  <reference id="jsonProcessor" interface="org.apache.camel.Processor" filter="(name=json)"/>  <cxf:rsClient id="rsClient" address="http://localhost:8181/cxf/test/"/>  <camelContext xmlns="http://camel.apache.org/schema/blueprint">     <route id="first">        <from uri="jetty:http://0.0.0.0:9090/first"/>        <convertBodyTo type="java.lang.String"/>        <wireTap uri=jms:queue:second?connectionFactory=connectionFactory"/>        <setBody><constant>OK</constant></setBody>     </route>     <route id="second">       <from uri="jms:queue:second?connectionFactory=connectionFactory&amp;acknowledgementModeName=CLIENT_ACKNOWLEDGE&amp;cacheLevelName=CACHE_CONSUMER"/>       <process ref="jsonProcessor"/>       <setHeader headerName="operationName"><constant>updateResource</constant></setHeader>       <setHeader headerName="CamelCxfRsUsingHttpAPI"><constant>false</constant></setHeader>       <setHeader herderName="CamelAcceptContentType"><constant>application/json</constant></setHeader>       <to uri="cxfrs://bean://rsClient?synchronous=true"/>     </route>  </camelContext></blueprint>

Persistent Redelivery

By default, when you create a Camel route, Camel automatically enables the default error handler. This error handler is an interceptor on the Camel channel (the “wire” between two processors in the route), and reacts when an exchange contains an exception.

In the case, it stops the exchange going through (waiting the handled flag) and retry the exchange.

Here, we wants:

  1. persist the message to be able to resume even in case of a platform restart or failure
  2. actually remove the message from the JMS queue only when it has been processed successfully
  3. replay the message for a number of times or infinite

Point 1 is already cover by default in Camel: by default, Camel will set the JMS messages as persistent. So, it means that the messages are stored in the ActiveMQ kahadb persistent store. Even if we restart ActiveMQ, those messages are not lost.

For point 2, we change the JMS acknowledgement mode. By default, Camel uses auto ack: it means that the ack is sent to the broker as soon as it’s consumed, so just after the jms endpoint in the Camel route. If a failure/exception occurs in the Camel route, then you have to deal with that, the message is already removed from the queue.
Changing the ack mode to client ack allows us to send the ack only at the OnCompletion time of the exchange. So the message will be removed from the queue only if it has been processed completely and succesfully. To do so, we add the ack mode on the jms endpoint URI:

<from uri="jms:queue:second?connectionFactory=connectionFactory&amp;acknowledgementModeName=CLIENT_ACKNOWLEDGE"/>

For point 3, we are going to change the provider URL in the ActiveMQ connection factory. By default, the ActiveMQ connection factory will redelivery a message 7 times max.
The number of redelivery attempt can be changed on the URL used by connection factory. For instance, to define a max number of redeliveries to 10, you can do:

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">        <property name="brokerURL" value="tcp://localhost:61616?jms.redeliveryPolicy.maximumRedeliveries=10" />    </bean>

You can also redelivery for ever (actually up to the expiration date of the message), using -1 for maximumRedeliveries:

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">        <property name="brokerURL" value="tcp://localhost:61616?jms.redeliveryPolicy.maximumRedeliveries=-1" />    </bean>

ActiveMQ Backoff

In the previous section, we used the ActiveMQ redelivery policy mechanism. That’s pretty convenient, but in order to optimize the number of attempts, we may want to introduce some delay between redeliveries. And more we have redeliveries, more we want to increase this delay. It’s what we call backoff redelivery.

This also can be configured on the URL of the connection factory using:

  • useExponentialBackOff enables an exponential backoff. It’s disabled by default.
  • backOffMultiplier defines how we increase the delay. By default, the “new” delay is 5 times longer than the “previous” one.
  • initialRedeliveryDelay is the early first redelivery delay. It’s the “startup” delay of the backoff policy. By default it’s 1000L, meaning 1 second.
  • maximumRedeliveryDelay is the max delay we can have. By default, it’s -1,
    meaning there’s not max.

So, let’s update the connection factory URL to enable backoff redelivery:

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">        <property name="brokerURL" value="tcp://localhost:61616?jms.redeliveryPolicy.maximumRedeliveries=-1&amp;jms.redeliveryPolicy.useExponentialBackOff=true&amp;jms.redeliveryPolicy.initialRedeliveryDelay=2000L&amp;jms.redeliveryPolicy.backOffMultiplier=2" />    </bean>

Here we have a redelivery policy with backoff that will double the delay for each attempt, starting from 2 seconds.

Thanks to that, we keep our Camel route pretty simple, we have consistent & persistent redelivery, using backoff policy.

Comments

Post a Comment

Popular posts from this blog

Quarkus and "meta" extension

Quarkus is a great Java framework with a large ecosystem thanks to the extensions . Basically, if using Maven, your pom.xml contains the list of the extensions you want to use, meaning something like this: ... <dependency> <groupId>io.quarkus</groupId> <artifactId>quarkus-grpc</artifactId> </dependency> <dependency> <groupId>io.quarkus</groupId> <artifactId>quarkus-opentelemetry</artifactId> </dependency> ... It can quickly become large, and sometime you can have "conflict" between extensions (for example if you mix in the same dependencies set reactive and non reactive extensions). Quarkus "meta" extension To simplify the dependencies for the extensions you want to use (especially if it's basically the same extensions in all your services), you can create your own "meta" extension. This extension doesn't contain actual code, it "wraps" other exten
Read more

Getting started with Apache Karaf Minho

Image
Apache Karaf Minho Apache Karaf Minho is a new runtime, completely created from scratch. The objectives are: super easy to use and run ligtning fast, including GraalVM support Kubernetes compliant to be easily used on the cloud helping to optimize cloud costs by supporting applications colocation providing turnkey services allowing you to create Minho applications in a minute Minho is composed by: minho-boot is the core of Minho. It contains the runtime launcher and the core services (service registry, lifecycle service, configuration service, ...). minho-* services provided additional functionnalities in the runtime. You can use these services to create your own services, it's also where you will find the services managing sepcific kind of applications, allowing colocation. minho-tooling optionally provide convenient tools to create your own runtime. It includes minho-maven-plugin for instance. In this blog post, we will take a quick tour on A
Read more

Apache Karaf Minho and OpenTelemetry

Image
Telemetry is an important part of a software stack, especially on distributed system (where services call each others). OpenTelemetry is a CNCF project coming from OpenTracing and OpenCensus. It provides API to easily instrument your code and tools to collect and export telemetry data (metrics, logs and traces). Before going into how to use OpenTelemetry with Minho, we have to understand some key concepts: Span is a single building block representing an operation or some unit of work that you want to capture. They are capable of standing on their own, referencing (or following) from other spans, storing metadata, tags, etc. Trace is a visualisation of the life of a request (or series of other operations) that is made up of one or more spans. This collection of spans work together to paint a picture of the overall request, allowing you to easily reconstruct what happened within each span, etc. SpanContext is a wrapper of key values pairs that associates a trace to one or mor
Read more