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

Event driven programming using Spring Boot and Reactor Example

Core concepts

Reactor:

Reactor is a framework to make event driven programming much easier and it is based on Reactor Design Pattern. Reactor is good for asynchronous applications on the JVM, it is an event gateway where event consumers are registered with a notification key.

Selector:

Selector is an abstraction to find consumer by invoking event.

Consumers and Event:

Consumers and Events as core module, Consumer is event consumer which needs to be notified for the event.

Producer:

Producer produces the Events and publish.

Here is one example which use the Reactor pattern to Message produce and consumes.

1. Create a Maven project and add below dependencies

<properties>
        <java.version>1.8</java.version>
        <version.reactor>2.0.6.RELEASE</version.reactor>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bus</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>org.reactivestreams</groupId>
            <artifactId>reactive-streams</artifactId>
            <version>1.0.0.final</version>
        </dependency>

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-stream</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-groovy</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.spring</groupId>
            <artifactId>reactor-spring-core</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.spring</groupId>
            <artifactId>reactor-spring-context</artifactId>
            <version>${version.reactor}</version>
        </dependency>

    </dependencies>

2. Create Customer POJO

package com.vinod.test;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Customer {
    private String name;
    private String address;
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getAddress() {
        return address;
    }
    public void setAddress(String address) {
        this.address = address;
    }
    @Override
    public String toString() {
        return "Customer [name=" + name + ", address=" + address + "]";
    }

}

3. Create a Consumer

package com.vinod.test;
import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.fn.Consumer;

@Service
class CustomerReceiver implements Consumer<Event<Customer>> {

    public void accept(Event<Customer> ev) {
        System.out.println("Customer " + ev.getData());
    }

}
 

4. Create a publisher

package com.vinod.test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.bus.EventBus;

@Service
public class CustomerPublisher {

    @Autowired
    EventBus eventBus;

    public void publishCustomerDetails() throws InterruptedException {
        Customer customer = new Customer();
        customer.setName("vinod");
        customer.setAddress("Sasi area");
        eventBus.notify("customer", Event.wrap(customer));
        System.out.println("Message sent");
    }

}

5. Spring boot Main class

package com.vinod.test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import reactor.Environment;
import reactor.bus.EventBus;

import static reactor.bus.selector.Selectors.$;

@Configuration
@EnableAutoConfiguration
@ComponentScan
public class Application implements CommandLineRunner {

    @Bean
    Environment env() {
        return Environment.initializeIfEmpty().assignErrorJournal();
    }

    @Bean
    EventBus createEventBus(Environment env) {
        return EventBus.create(env, Environment.THREAD_POOL);
    }

    @Autowired
    private EventBus eventBus;
    @Autowired
    private CustomerReceiver customerReceiver;
    @Autowired
    private CustomerPublisher customerPublisher;
    @Autowired
    private AdminReceiver adminReceiver;

    public void run(String... args) throws Exception {
        eventBus.on($("customer"), customerReceiver);
        customerPublisher.publishCustomerDetails();
    }

    public static void main(String[] args) throws InterruptedException {
        SpringApplication.run(Application.class, args);

    }

}

6. Run the program (Application.java)

Here we can see the publisher sends the message and consumer is consuming the same.

Message sent
Customer Customer [name=vinod, address=Sasi area]

7. Download Example

https://github.com/kkvinodkumaran/spring

Java Drools Example

Drools is a open source Business Logic integration Platform based on java from JBoss and RedHat. Here is one simple example to execute business logics based on rules. see more about drools

1. Create a maven project and add below dependencies

 
  <dependencies>
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-compiler</artifactId>
            <version>6.0.0.CR1</version>
        </dependency>
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-core</artifactId>
            <version>6.0.0.CR1</version>
        </dependency>
    </dependencies>

2. Create a .drl file to add our business logic (eligibility.drl)

import com.vinod.test.Customer
 
dialect "mvel"
 
rule "Customer eligibility rule"
    when
        $customer : Customer(age>=21)              
    then
        System.out.println($customer.name" Eligibile for benefits");
end

3. Create Customer pojo class

package com.vinod.test;

public class Customer {
    private String name;
    private int age;
    public Customer() {
    }
    public Customer(String name, int age) {
        this.name = name;
        this.age = age;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public int getAge() {
        return age;
    }
    public void setAge(int age) {
        this.age = age;
    }
}
 

4. Create a main class to load and fire rules

package com.vinod.test;

import java.io.File;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.List;

import org.kie.api.KieServices;
import org.kie.api.builder.KieBuilder;
import org.kie.api.builder.KieFileSystem;
import org.kie.api.builder.KieRepository;
import org.kie.api.io.KieResources;
import org.kie.api.io.Resource;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;

public class CustomerEligibilityRuleTest {

    public static void main(String[] args) {
        List<File> rules = new ArrayList<File>();
        File currentdir = new File(System.getProperty("user.dir"));
        File[] listFiles = currentdir.listFiles(new FilenameFilter() {
            public boolean accept(File dir, String name) {
                return (name.endsWith(".drl"));
            }
        });
        for (File listFile : listFiles) {
            System.out.println(listFile.getName());
            rules.add(listFile);
        }
        List<Customer> customers = new ArrayList<Customer>();
        customers.add(new Customer("Vinod k kumaran", 21));
        executeRules(rules, customers);
    }

    public static void executeRules(List<File> files, List<Customer> customers) {
        try {
            KieServices kieServices = KieServices.Factory.get();
            KieResources kieResources = kieServices.getResources();
            KieFileSystem kieFileSystem = kieServices.newKieFileSystem();
            KieRepository kieRepository = kieServices.getRepository();
            for (File ruleFile : files) {
                Resource resource = kieResources.newFileSystemResource(ruleFile);
                kieFileSystem.write(resource);
            }
            KieBuilder kb = kieServices.newKieBuilder(kieFileSystem);
            kb.buildAll();
            KieContainer kContainer = kieServices.newKieContainer(kieRepository.getDefaultReleaseId());
            KieSession kSession = kContainer.newKieSession();
            kSession.addEventListener(new CustomerRuleTracker());
            for (Object customer : customers) {
                kSession.insert(customer);
            }
            kSession.fireAllRules();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 

5. Add rule tracker class to listen all stages of firing rules 

This class is added in the KieSession already in the above main class

package com.vinod.test;

import org.drools.core.event.BeforeActivationFiredEvent;
import org.kie.api.event.rule.AfterMatchFiredEvent;
import org.kie.api.event.rule.BeforeMatchFiredEvent;
import org.kie.api.event.rule.DefaultAgendaEventListener;
import org.kie.api.event.rule.MatchCancelledEvent;
import org.kie.api.event.rule.MatchCreatedEvent;
import org.kie.api.event.rule.MatchEvent;
import org.kie.api.event.rule.RuleFlowGroupActivatedEvent;

public class CustomerRuleTracker extends DefaultAgendaEventListener {

    @Override
    public void afterMatchFired(AfterMatchFiredEvent event) {
        System.out.println("Executing afterMatchFired ");
    }
    @Override
    public void beforeRuleFlowGroupActivated(RuleFlowGroupActivatedEvent event) {
        System.out.println("Executing beforeRuleFlowGroupActivated ");

    }
    @Override
    public void beforeMatchFired(BeforeMatchFiredEvent event) {
        System.out.println("Executing beforeMatchFired ");
    }
    public void beforeActivationFired(BeforeActivationFiredEvent event) {
        System.out.println("Executing beforeActivationFired ");

    }
    @Override
    public void matchCreated(MatchCreatedEvent event) {
        System.out.println("Executing matchCreated ");
        registerActions(event);
    }
    @Override
    public void matchCancelled(MatchCancelledEvent event) {
        System.out.println("Executing matchCancelled");

    }
    private void registerActions(MatchEvent event) {
        System.out.println("Executing registerActions");
    }
}

6. Run the main program.

We can see the customer name + eligible for benefits printed as per our rules..

eligibility.drl

Executing matchCreated
Executing registerActions
Executing beforeMatchFired
Vinod k kumaran Eligibile for benefits
Executing afterMatchFired

7. Download Example

https://github.com/kkvinodkumaran/drools

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