Apache Camel is a rules-based routing and arbitration engine that provides a Java object-based implementation of the Enterprise Integration Pattern, using an API (or Declarative Java Domain Specific Language) to configure routing and arbitration rules. We can implement exception handling in two ways Using Do Try block and OnException block . A re-try policy defines rules when Camel Error Handler perform re-try attempts. e.g you can setup rules that state how many times to try retry, and the delay in between attempts, and so forth.
The project structure will be as follows-
The pom.xml will be as follows-
<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/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.spring</groupId> <artifactId>camel-spring-integration</artifactId> <version>0.0.1-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-core</artifactId> <version>2.13.0</version> </dependency> <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-spring</artifactId> <version>2.13.0</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.5</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.5</version> </dependency> </dependencies> </project>
Added MainApplication class
package com.spring.main;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class MainApplication {
/*
* It is a Main Application. It invokes routeBuilder bean through from applicationContect.xml file via
* ClassPathXmlApplicationContext
*/
public static void main(String[] args) {
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
ctx.start();
System.out.println("Application started...");
try {
System.out.println("inside try block");
System.out.println("--------------------- inputMessageBody ------------------- ");
Thread.sleep(5 * 60 * 1000);
}
catch (InterruptedException e) {
e.printStackTrace();
}
ctx.stop();
ctx.close();
}
}
Created CamelCustomException class to implement custom exception
package com.spring.exception;
/*
* Created the custom exception...
*/
public class CamelCustomException extends Exception {
private static final long serialVersionUID = 2L;
}
Created applicationContext.xml
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd"> <bean id="routeBuilder" class="com.spring.route.SimpleRouteBuilder" /> <camelContext xmlns="http://camel.apache.org/schema/spring"> <routeBuilder ref="routeBuilder" /> </camelContext> </beans>
We modify the MyProcessor as follows. We check if the input contains text “test”, then an custom exception is thrown.
package com.spring.processor;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import com.spring.exception.CamelCustomException;
public class MyProcessor implements Processor {
/*
* (non-Javadoc)
* @see org.apache.camel.Processor#process(org.apache.camel.Exchange)
* It is Camel processor class which implements Processor.
*/
public void process(Exchange exchng) throws Exception {
String inputMessageBody = exchange.getIn().getBody(String.class);
System.out.println("\n" + inputMessageBody);
if (inputMessageBody.contains("test"))
throw new CamelCustomException();
}
}
Our SimpleRoutebuilder class is as before-
package com.spring.route;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import com.spring.exception.CamelCustomException;
import com.spring.processor.MyProcessor;
import com.spring.processor.RetryProcessor;
public class SimpleRouteBuilder extends RouteBuilder {
/*
* (non-Javadoc)
* @see org.apache.camel.builder.RouteBuilder#configure()
* It is a SimpleRouteBuilder class which facilitates Routes
*/
@Override
public void configure() throws Exception {
onException(CamelCustomException.class).process(new Processor() {
public void process(Exchange exchng) throws Exception {
System.out.println("Exception is handling by onException{} block");
}
})
.log("Received body ${body}").handled(true);
from("file:/home/knoldus/Downloads/Softwares/Workspace/input?noop=true")
.process(new MyProcessor())
.to("file:/home/knoldus/Downloads/Softwares/Workspace/output");
}
}
After running the MainApp.java output will be as follow-
After the exception is thrown it is caught by the onException{} Block. We will now define the redelivery policy. So now each message will be tried 3 times before being caught.
Modify the applicationContext.xml
1.) maximumRedeliveries is the maximum number of times can we redeliver.
2.) redeliveryDelay is the delay time (in ms) between each re-try attempts.
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd"> <bean id="routeBuilder" class="com.spring.route.SimpleRouteBuilder" /> <camelContext xmlns="http://camel.apache.org/schema/spring"> <routeBuilder ref="routeBuilder" /> <redeliveryPolicyProfile id="localRedeliveryPolicyProfile" retryAttemptedLogLevel="WARN" maximumRedeliveries="3" redeliveryDelay="1" /> </camelContext> </beans>
Configure the redelivery policy in the route of SimpleRouteBuilder class. So that it can process the redelivery logic.
package com.spring.route;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import com.spring.exception.CamelCustomException;
import com.spring.processor.MyProcessor;
import com.spring.processor.RetryProcessor;
public class SimpleRouteBuilder extends RouteBuilder {
/*
* (non-Javadoc)
* @see org.apache.camel.builder.RouteBuilder#configure()
* It is a SimpleRouteBuilder class which facilitates Routes
*/
@Override
public void configure() throws Exception {
onException(CamelCustomException.class).process(new Processor() {
public void process(Exchange exchange) throws Exception {
System.out.println("Exception is handling by onException{} block");
}
})
.redeliveryPolicyRef("localRedeliveryPolicyProfile")
.log("Received body ${body}").handled(true);
from("file:/home/knoldus/Downloads/Softwares/Workspace/input?noop=true")
.process(new MyProcessor())
.to("file:/home/knoldus/Downloads/Softwares/Workspace/output");
}
}
After running the MainApp.java output will be as follows-
Create RetryProcessor class. It will responsible for process the logic in retry attempts.
package com.spring.processor;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
/**
* @author knoldus
* If the body contains text "test" then it will set new body
*/
public class RetryProcessor implements Processor {
public void process(Exchange exchange) throws Exception {
exchange.getIn().setBody("replaced new body...");
}
}
Configure the onRedelivery before the redelivery policy configuration in the route of SimpleRouteBuilder class
package com.spring.route;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import com.spring.exception.CamelCustomException;
import com.spring.processor.MyProcessor;
import com.spring.processor.RetryProcessor;
public class SimpleRouteBuilder extends RouteBuilder {
/*
* (non-Javadoc)
* @see org.apache.camel.builder.RouteBuilder#configure()
* It is a SimpleRouteBuilder class which facilitates Routes
*/
@Override
public void configure() throws Exception {
onException(CamelCustomException.class).process(new Processor() {
public void process(Exchange exchange) throws Exception {
System.out.println("Exception is handling by onException{} block");
}
})
.onRedelivery(new RetryProcessor())
.redeliveryPolicyRef("localRedeliveryPolicyProfile")
.log("Received body ${body}").handled(true);
from("file:/home/knoldus/Downloads/Softwares/Workspace/input?noop=true")
.process(new MyProcessor())
.to("file:/home/knoldus/Downloads/Softwares/Workspace/output");
}
}
In the Re-try Policy bean we have changed the exchange data so the exception will not thrown again
Conclusion
In this blog, we have covered how to implement and configure Exception Handling with retry policy in Apache Camel using Spring. Now you are ready to go to implement Exception Handling with retry policy in Apache Camel using Spring. For more, you can refer to the documentation: https://people.apache.org/~dkulp/camel/redeliverypolicy.html
