Added Apache Kafka component

components/camel-kafka/pom.xml

@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+-->
+<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>
+
+    <parent>
+        <groupId>org.apache.camel</groupId>
+        <artifactId>components</artifactId>
+        <version>2.13-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>camel-kafka</artifactId>
+    <packaging>bundle</packaging>
+    <name>Camel :: Kafka</name>
+    <description>Camel kafka support</description>
+
+    <properties>
+        <camel.osgi.export.pkg>org.apache.camel.component.kafka.*</camel.osgi.export.pkg>
+        <kafka.version>0.8.0</kafka.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.10</artifactId>
+            <version>${kafka.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-library</artifactId>
+            <version>${scala-version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java

@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultComponent;
+
+/**
+ * @author Stephen Samuel
+ */
+public class KafkaComponent extends DefaultComponent {
+
+    public KafkaComponent() {
+    }
+
+    public KafkaComponent(CamelContext context) {
+        super(context);
+    }
+
+    @Override
+    protected KafkaEndpoint createEndpoint(String uri,
+                                           String remaining,
+                                           Map<String, Object> params) throws Exception {
+        KafkaEndpoint endpoint = new KafkaEndpoint(uri, remaining, this);
+        setProperties(endpoint, params);
+        return endpoint;
+    }
+}

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java

@@ -0,0 +1,30 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka;
+
+/**
+ * @author Stephen Samuel
+ */
+public class KafkaConstants {
+
+    public static final String DEFAULT_GROUP = "group1";
+
+    public static final String PARTITION_KEY = "kafka.PARTITION_KEY";
+    public static final String PARTITION = "kafka.EXCHANGE_NAME";
+    public static final String KEY = "kafka.CONTENT_TYPE";
+    public static final String TOPIC = "kafka.TOPIC";
+}

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java

@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.DefaultConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import kafka.consumer.ConsumerConfig;
+import kafka.consumer.ConsumerIterator;
+import kafka.consumer.KafkaStream;
+import kafka.javaapi.consumer.ConsumerConnector;
+import kafka.message.MessageAndMetadata;
+
+/**
+ * @author Stephen Samuel
+ */
+public class KafkaConsumer extends DefaultConsumer {
+
+    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
+
+    private final KafkaEndpoint endpoint;
+    private final Processor processor;
+
+    ConsumerConnector consumer;
+    ExecutorService executor;
+
+    public KafkaConsumer(KafkaEndpoint endpoint, Processor processor) {
+        super(endpoint, processor);
+        this.endpoint = endpoint;
+        this.processor = processor;
+        if (endpoint.getZookeeperHost() == null)
+            throw new IllegalArgumentException("zookeeper host must be specified");
+        if (endpoint.getZookeeperPort() == 0)
+            throw new IllegalArgumentException("zookeeper port must be specified");
+        if (endpoint.getGroupId() == null)
+            throw new IllegalArgumentException("groupId must not be null");
+    }
+
+    Properties getProps() {
+        Properties props = new Properties();
+        props.put("zookeeper.connect", endpoint.getZookeeperHost() + ":" + endpoint.getZookeeperPort());
+        props.put("group.id", endpoint.getGroupId());
+        props.put("zookeeper.session.timeout.ms", "400");
+        props.put("zookeeper.sync.time.ms", "200");
+        props.put("auto.commit.interval.ms", "1000");
+        return props;
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+        log.info("Starting Kafka consumer");
+
+        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(new ConsumerConfig(getProps()));
+
+        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
+        topicCountMap.put(endpoint.getTopic(), endpoint.getConsumerStreams());
+        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
+        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(endpoint.getTopic());
+
+        executor = endpoint.createExecutor();
+        for (final KafkaStream stream : streams) {
+            executor.submit(new ConsumerTask(stream));
+        }
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        super.doStop();
+        log.info("Stopping Kafka consumer");
+
+        if (consumer != null)
+            consumer.shutdown();
+        if (executor != null)
+            executor.shutdown();
+        executor = null;
+    }
+
+    class ConsumerTask implements Runnable {
+
+        private KafkaStream stream;
+
+        public ConsumerTask(KafkaStream stream) {
+            this.stream = stream;
+        }
+
+        public void run() {
+            ConsumerIterator<byte[], byte[]> it = stream.iterator();
+            while ((Boolean) it.hasNext()) {
+                MessageAndMetadata<byte[], byte[]> mm = it.next();
+                Exchange exchange = endpoint.createKafkaExchange(mm);
+                try {
+                    processor.process(exchange);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+}
+