Apache Camel Kafka Component Example

Apache camel API has the inbuilt kafka component and it is very simple to create producer, consumer and process messages. Here is one simple Kafka producer and consumer example using Apache camel and Kafka.

Steps

1) Download Apache kafka from https://kafka.apache.org/downloads

2) Extract the tar and start the zookeeper and kafka 

 ./zookeeper-server-start.sh ../config/zookeeper.properties

 ./kafka-server-start.sh ../config/server.properties

3. Create a Maven project and add below dependencies

<dependencies>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-core</artifactId>
			<version>2.16.3</version>
		</dependency>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-kafka</artifactId>
			<version>2.16.3</version>
		</dependency>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-jetty</artifactId>
			<version>2.16.3</version>
		</dependency>
		<dependency>
			<groupId>org.scala-lang</groupId>
			<artifactId>scala-library</artifactId>
			<version>2.11.0</version>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
			<version>1.6.6</version>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>1.6.6</version>
		</dependency>
		<dependency>
			<groupId>log4j</groupId>
			<artifactId>log4j</artifactId>
			<version>1.2.17</version>
		</dependency>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-test</artifactId>
			<version>2.17.3</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>com.fasterxml.jackson.core</groupId>
			<artifactId>jackson-databind</artifactId>
		</dependency>
	</dependencies>

4. Create a camel main class to add camel routes and its processors.

package com.vinod.test;

import org.apache.camel.Exchange;
import org.apache.camel.Main;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;

public class CamelKafkaConsumerTest {

    public static void main(String... args) throws Exception {
	Main main = new Main();
	main.enableHangupSupport();
	main.addRouteBuilder(new MyCamelJettyBuilder());
	main.run(args);
    }
}

class MyCamelJettyBuilder extends RouteBuilder {
    String topicName = "topic=test";
    String kafkaServer = "kafka:localhost:9092";
    String zooKeeperHost = "zookeeperHost=localhost&zookeeperPort=2181";
    String serializerClass = "serializerClass=kafka.serializer.StringEncoder";
    String autoOffsetOption = "autoOffsetReset=smallest";
    String groupId = "groupId=testingvinod";

    String toKafka = new StringBuilder().append(kafkaServer).append("?").append(
	    topicName).append("&").append(zooKeeperHost).append("&").append(
		    serializerClass).toString();

    String fromKafka = new StringBuilder().append(toKafka).append("&").append(
	    autoOffsetOption).append("&").append(groupId).toString();

    public void configure() {
	from("jetty:http://localhost:8182/mytestservice").process(
		new Processor() {
		    public void process(Exchange exchange) throws Exception {
			String message = exchange.getIn().getBody(String.class);
			exchange.getIn().setBody(message, String.class);
			exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY,
				0);
			exchange.getIn().setHeader(KafkaConstants.KEY, "1");
		    }
		}).to(toKafka);
	from(fromKafka).process(new Processor() {
	    public void process(Exchange exchange) throws Exception {
		if (exchange.getIn() != null) {
		    Message message = exchange.getIn();
		    String data = message.getBody(String.class);
		    System.out.println("Data =" + data.toString());
		}
	    }
	});

    }
}

In this example exposed a jetty end point to give the input message instead of static message, once started the above program use any rest client to test the program.

5) Send message to Kafka topic using our service .. http://localhost:8182/mytestservice

now we can see the output which we sent and consumed from Kafka in the console

Data  =Test my message by Vinod

6) Download Example

https://github.com/kkvinodkumaran/camel

Reference: Apache Camel, Kafka

PGP Encryption and Decryption using Apache Camel

In this example we will see how do to the file encryption and decryption using Apache Camel using pgp

1. Generate pgp keys

In order to do the encryption/decryption we required the pgp public and private keys, in this example we will use below portal to generate the keys. Once we created the files we need to place it in to our source resource folder

https://www.igolder.com/pgp/generate-key/

 

2. Create a maven project and add below dependencies

<properties>
        <camel.version>2.13.2</camel.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-ftp</artifactId>
            <version>2.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-core</artifactId>
            <version>${camel.version}</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-crypto</artifactId>
            <version>${camel.version}</version>

        </dependency>
        <dependency>
            <groupId>bouncycastle</groupId>
            <artifactId>bcpg-jdk13</artifactId>
            <version>121</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>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
    </dependencies>

3. Create a Camel Route to start encryption

 

package com.vinod.test;

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.converter.crypto.PGPDataFormat;
import org.apache.camel.main.Main;

public class EncryptionTest {

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Main main = new Main();
        main.enableHangupSupport();
        main.addRouteBuilder(new MyRouteBuilder());
        main.run(args);

    }

}

class MyRouteBuilder extends RouteBuilder {

    @Override
    public void configure() {

        try {
            System.out.println("My Encryption/Decryption route started");
            // Encryption
            PGPDataFormat encrypt = new PGPDataFormat();
            encrypt.setKeyFileName("vinodpublickeyfile.pgp");
            encrypt.setKeyUserid("vinod");

            from("file:tobeencrypt?noop=true;delete=true").marshal(encrypt).to("file:encrypted");

            // Decryption

            PGPDataFormat decrypt = new PGPDataFormat();
            decrypt.setKeyFileName("vinodprivatekeyfile.pgp");

            decrypt.setKeyUserid("vinod");
            decrypt.setPassword("vinod@123");

            from("file:tobedecrypt").unmarshal(decrypt).to("file:decrypted");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

5. Run the program

After running the above main program camel route will start and we can see the two directories are created and we can place the files which we needs to be encrypted.

Once we placed the file in to tobeencrypt directory Camel route will process that file and place in to encrypted directory after encryption

 

 

 

 

 

 

 

 

 

 

 

 

Download example

https://github.com/kkvinodkumaran/camel 

 

 

 

 

Apache Camel Mongodb component Example

Apache camel provides many components to interact with external systems, here is one simple example to connect mongodb using from apache camel route.

Create a maven project and add below dependencies

<dependencies>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-core</artifactId>
            <version>2.12.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring</artifactId>
            <version>2.12.1</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>3.2.4.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-beans</artifactId>
            <version>3.2.4.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>3.2.4.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
            <version>3.2.4.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-jetty</artifactId>
            <version>2.12.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-mongodb</artifactId>
            <version>2.12.1</version>
        </dependency>
    </dependencies>

 

Create a Camel route

package com.vinod.test;

import org.apache.camel.builder.RouteBuilder;

public class CamelMongoRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        from("jetty:http://localhost:8181/mongoSelect")
                .to("mongodb:myDb?database=customerdb&collection=customer&operation=findAll");
        from("jetty:http://localhost:8181/mongoInsert")
                .to("mongodb:myDb?database=customerdb&collection=customer&operation=insert");
    }
}

Create application context.xml

<?xml version="1.0" encoding="UTF-8"?>
<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">
    <camel:camelContext id="camel-client">
        <camel:routeBuilder ref="vinodroute" />
    </camel:camelContext>
    <bean id="myDb" class="com.mongodb.Mongo">
        <constructor-arg index="0" value="localhost" />
    </bean>
    <bean id="vinodroute" class="com.vinod.test.CamelMongoRoute" />
</beans>

Create a Main program

This main program will start the routes

package com.vinod.test;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class CamelMongoMain {
    @SuppressWarnings("resource")
    public static void main(String[] args) {
                      new ClassPathXmlApplicationContext("application-context.xml");
    }

}
 

Run it

1. Start Mongodb
2. Run the main program
3. Use rest client to give inputs

To insert data into mongodb

Url= http://localhost:8181/mongoInsert

Request body= {name:"Vinod K K"}

Retrieve data
Url= http://localhost:8181/mongoSelect

Response from mongodb = [{ "_id" : { "$oid" : "5419cefb0364771a127e352c"} , "name" : "Vinod K K"}]

Data in mongodb
> use customerdb
switched to db customerdb
> show collections
customer
system.indexes
> db.customer.find()
{ "_id" : ObjectId("5419cefb0364771a127e352c"), "name" : "Vinod K K" }
> 

Done !!!!

Download example

https://github.com/kkvinodkumaran/camel

Camel Timer Example

Camel Timer component is used to generate or process message exchanges when a time fires. Here is one example to print ‘Hello world ‘ in each five seconds

Example

package com.vinod.test;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.main.Main;
public class TimerTest {
    public static void main(String... args) throws Exception {
        Main main = new Main();
        main.enableHangupSupport();
        main.addRouteBuilder(new TimerRoute());
        main.run(args);
    }
}

class TimerRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        from("timer://foo?period=5000").process(new Processor() {
            public void process(Exchange exchange) throws Exception {
                System.out.println("Hello world  :"
                        + System.currentTimeMillis());
            }
        });
    }
}

Output

Hello world  :1453013393616

Hello world  :1453013398609

Hello world  :1453013403614

Hello world  :1453013408617

Hello world  :1453013413621

Hello world  :1453013418624

 

Done!! download source codes from

 

https://github.com/kkvinodkumaran/camel

Ref: http://camel.apache.org/timer.html

Camel Exception clause example

Camel provides Exception clause to handle exceptions , came RouteBuilder onException method will helps to handle exceptions. In this example we will move files from source directory to destination and in case RuntimeException occured we will move the messages to exceptionDirectory.

package com.vinod.test;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.main.Main;

public class OnExceptionTest {
    public static void main(String... args) throws Exception {
        Main main = new Main();
        main.enableHangupSupport();
        main.addRouteBuilder(new TestExceptionRoute());
        main.run(args);
    }
}

class TestExceptionRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        onException(RuntimeException.class).to("file:exceptionDirectory");
        from("file:source").process(new Processor() {
            public void process(Exchange exchange) throws Exception {
                throw new RuntimeException();
            }
        }).to("file:destination");
    }

}
 

Download source code 

https://github.com/kkvinodkumaran/camel

 

 

 

Camel DeadLetterChannel errorhandler Example

Camel supports the Dead Letter Channel from EIP patterns to handle errors. Camel will move the Exchange data in case any exceptions to Dead Letter channel. In this example we will see Route Builder error handler is moving the exchange data to dead letter channel. For testing this While moving the files from source to destination we are throwing exception.

package com.vinod.test;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.main.Main;
public class ErrorHandlerTest {
    public static void main(String... args) throws Exception {
        Main main = new Main();
        main.enableHangupSupport();
        main.addRouteBuilder(new TestRoute());
        main.run(args);
    }
}

class TestRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        errorHandler(deadLetterChannel("file:deadletter"));
        from("file:source").process(new Processor() {
            public void process(Exchange exchange) throws Exception {
                throw new RuntimeException();
            }
        }).to("file:destination");
    }
}

Camel Data transformation using Custom Expression

Camel provides custom expression class to control the data transformation. Here is one example which will convert the text file to html format.

Example

package com.vinod.test;

import org.apache.camel.Exchange;
import org.apache.camel.Expression;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.main.Main;

public class CamelDatatransformation {
    public static void main(String... args) throws Exception {
        Main main = new Main();
        main.enableHangupSupport();
        main.addRouteBuilder(new MyRoute());
        main.run(args);
    }
}

class MyRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        from("file:source").transform(new Expression() {
            public <T> T evaluate(Exchange exchange, Class<T> type) {
                String body = exchange.getIn().getBody(String.class);
                body = body.replaceAll("\n", "<br/>");
                body = "<html><body>" + body + "</body></html>";
                return (T) body;
            }
        }).to("file:destination");
    }
}

Run it and place one input text file in source folder and see the destination folder

Done!!!

Camel Jetty Component Example

The jetty component provides HTTP-based endpoints for consuming and producing HTTP requests. Here is one simple example to expose a jetty end point and serving the http request from REST client.

1. Create Maven project with below dependencies.

    <dependencies>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-jetty</artifactId>
            <version>2.12.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-core</artifactId>
            <version>2.12.1</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.6.6</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.6.6</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-test</artifactId>
            <version>2.12.1</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

2. Create a Simple Route to expose Jetty endpoint.

package com.vinod.test;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.main.Main;

public class CamelJettyExample {
    public static void main(String... args) throws Exception {
        Main main = new Main();
        main.enableHangupSupport();
        main.addRouteBuilder(new MyCamelJettyBuilder());
        main.run(args);
    }
}

class MyCamelJettyBuilder extends RouteBuilder {
    public void configure() {
        from("jetty:http://localhost:8181/mytestservice").process(new Processor() {
            public void process(Exchange exchange) throws Exception {
                String message = exchange.getIn().getBody(String.class);
                System.out.println("Hello Mr :" + message);
                exchange.getOut().setBody("Hello world Mr " + message);
            }
        });
    }
}

3. Run it
Run the application and hit the service
http://localhost:8181/mytestservice

4. Done !! Download application
https://github.com/kkvinodkumaran/camel