ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [15/19] incubator-ignite git commit: ignite-428 Implement IgniteKafkaStreamer to stream data from Apache Kafka
Date Mon, 06 Jul 2015 07:55:07 GMT
ignite-428 Implement IgniteKafkaStreamer to stream data from Apache Kafka


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/2c41739d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/2c41739d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/2c41739d

Branch: refs/heads/ignite-1086
Commit: 2c41739dcd83751270b6bd30c6f2595edc68a1b1
Parents: 9f6a7f9
Author: vishal.garg <vishal.garg@workday.com>
Authored: Mon Jun 22 19:35:08 2015 -0700
Committer: agura <agura@gridgain.com>
Committed: Fri Jul 3 19:39:11 2015 +0300

----------------------------------------------------------------------
 modules/kafka/pom.xml                           | 128 +++++++
 .../ignite/stream/kafka/KafkaStreamer.java      | 179 +++++++++
 .../stream/kafka/KafkaEmbeddedBroker.java       | 373 +++++++++++++++++++
 .../kafka/KafkaIgniteStreamerSelfTest.java      | 230 ++++++++++++
 .../ignite/stream/kafka/SimplePartitioner.java  |  46 +++
 pom.xml                                         |   1 +
 6 files changed, 957 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2c41739d/modules/kafka/pom.xml
----------------------------------------------------------------------
diff --git a/modules/kafka/pom.xml b/modules/kafka/pom.xml
new file mode 100644
index 0000000..165ec1c
--- /dev/null
+++ b/modules/kafka/pom.xml
@@ -0,0 +1,128 @@
+<?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.
+-->
+
+<!--
+    POM file.
+-->
+<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>
+
+    <parent>
+        <groupId>org.apache.ignite</groupId>
+        <artifactId>ignite-parent</artifactId>
+        <version>1</version>
+        <relativePath>../../parent</relativePath>
+    </parent>
+
+    <artifactId>ignite-kafka</artifactId>
+    <version>1.1.1-SNAPSHOT</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.10</artifactId>
+            <version>0.8.2.1</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.sun.jmx</groupId>
+                    <artifactId>jmxri</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.sun.jdmk</groupId>
+                    <artifactId>jmxtools</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>net.sf.jopt-simple</groupId>
+                    <artifactId>jopt-simple</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-simple</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.zookeeper</groupId>
+                    <artifactId>zookeeper</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <version>3.4.5</version>
+        </dependency>
+
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.4</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-log4j</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.ow2.asm</groupId>
+            <artifactId>asm-all</artifactId>
+            <version>4.2</version>
+        </dependency>
+
+
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.gridgain</groupId>
+            <artifactId>ignite-shmem</artifactId>
+            <version>1.0.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>commons-beanutils</groupId>
+            <artifactId>commons-beanutils</artifactId>
+            <version>1.8.3</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-spring</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-core</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2c41739d/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
new file mode 100644
index 0000000..e0240ce
--- /dev/null
+++ b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
@@ -0,0 +1,179 @@
+/*
+ * 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.ignite.stream.kafka;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.stream.*;
+
+import kafka.consumer.*;
+import kafka.javaapi.consumer.ConsumerConnector;
+import kafka.message.*;
+import kafka.serializer.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+/**
+ * Server that subscribes to topic messages from Kafka broker, streams its to key-value pairs
into {@link
+ * org.apache.ignite.IgniteDataStreamer} instance.
+ * <p>
+ * Uses Kafka's High Level Consumer API to read messages from Kafka
+ *
+ * @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example">Consumer
Consumer Group
+ * Example</a>
+ */
+public class KafkaStreamer<T, K, V>
+    extends StreamAdapter<T, K, V> {
+
+    /** Logger. */
+    private IgniteLogger log;
+
+    /** Executor used to submit kafka streams. */
+    private ExecutorService executor;
+
+    /** Topic. */
+    private String topic;
+
+    /** Number of threads to process kafka streams. */
+    private int threads;
+
+    /** Kafka Consumer Config. */
+    private ConsumerConfig consumerConfig;
+
+    /** Key Decoder. */
+    private Decoder<K> keyDecoder;
+
+    /** Value Decoder. */
+    private Decoder<V> valueDecoder;
+
+    /** Kafka Consumer connector. */
+    private ConsumerConnector consumer;
+
+    /**
+     * Sets the topic.
+     *
+     * @param topic Topic Name.
+     */
+    public void setTopic(final String topic) {
+        this.topic = topic;
+    }
+
+    /**
+     * Sets the threads.
+     *
+     * @param threads Number of Threads.
+     */
+    public void setThreads(final int threads) {
+        this.threads = threads;
+    }
+
+    /**
+     * Sets the consumer config.
+     *
+     * @param consumerConfig  Consumer configuration.
+     */
+    public void setConsumerConfig(final ConsumerConfig consumerConfig) {
+        this.consumerConfig = consumerConfig;
+    }
+
+    /**
+     * Sets the key decoder.
+     *
+     * @param keyDecoder Key Decoder.
+     */
+    public void setKeyDecoder(final Decoder<K> keyDecoder) {
+        this.keyDecoder = keyDecoder;
+    }
+
+    /**
+     * Sets the value decoder.
+     *
+     * @param valueDecoder Value Decoder
+     */
+    public void setValueDecoder(final Decoder<V> valueDecoder) {
+        this.valueDecoder = valueDecoder;
+    }
+
+    /**
+     * Starts streamer.
+     *
+     * @throws IgniteException If failed.
+     */
+    public void start() {
+        A.notNull(getStreamer(), "streamer");
+        A.notNull(getIgnite(), "ignite");
+        A.notNull(topic, "topic");
+        A.notNull(keyDecoder, "key decoder");
+        A.notNull(valueDecoder, "value decoder");
+        A.notNull(consumerConfig, "kafka consumer config");
+        A.ensure(threads > 0, "threads > 0");
+
+        log = getIgnite().log();
+
+        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);
+
+        Map<String, Integer> topicCountMap = new HashMap<>();
+        topicCountMap.put(topic, threads);
+
+        Map<String, List<KafkaStream<K, V>>> consumerMap = consumer.createMessageStreams(topicCountMap,
keyDecoder,
+            valueDecoder);
+
+        List<KafkaStream<K, V>> streams = consumerMap.get(topic);
+
+        // Now launch all the consumer threads.
+        executor = Executors.newFixedThreadPool(threads);
+
+        // Now create an object to consume the messages.
+        for (final KafkaStream<K,V> stream : streams) {
+            executor.submit(new Runnable() {
+                @Override public void run() {
+
+                    ConsumerIterator<K, V> it = stream.iterator();
+
+                    while (it.hasNext()) {
+                        final MessageAndMetadata<K, V> messageAndMetadata = it.next();
+                        getStreamer().addData(messageAndMetadata.key(), messageAndMetadata.message());
+                    }
+                }
+            });
+        }
+    }
+
+    /**
+     * Stops streamer.
+     */
+    public void stop() {
+        if (consumer != null)
+            consumer.shutdown();
+
+        if (executor != null) {
+            executor.shutdown();
+
+            try {
+                if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS))
+                    if (log.isDebugEnabled())
+                        log.debug("Timed out waiting for consumer threads to shut down, exiting
uncleanly");
+            }
+            catch (InterruptedException e) {
+                if (log.isDebugEnabled())
+                    log.debug("Interrupted during shutdown, exiting uncleanly");
+            }
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2c41739d/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
new file mode 100644
index 0000000..28533f7
--- /dev/null
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
@@ -0,0 +1,373 @@
+/*
+ * 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.ignite.stream.kafka;
+
+import org.apache.commons.io.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.zookeeper.server.*;
+
+import kafka.admin.*;
+import kafka.api.*;
+import kafka.api.Request;
+import kafka.producer.*;
+import kafka.server.*;
+import kafka.utils.*;
+import org.I0Itec.zkclient.*;
+
+import java.io.*;
+import java.net.*;
+import java.nio.file.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+/**
+ * Kafka Embedded Broker.
+ */
+public class KafkaEmbeddedBroker {
+
+    /** Default ZooKeeper Host. */
+    private static final String ZK_HOST = "localhost";
+
+    /** Broker Port. */
+    private static final int BROKER_PORT = 9092;
+
+    /** ZooKeeper Connection Timeout. */
+    private static final int ZK_CONNECTION_TIMEOUT = 6000;
+
+    /** ZooKeeper Session Timeout. */
+    private static final int ZK_SESSION_TIMEOUT = 6000;
+
+    /** ZooKeeper port. */
+    private static int zkPort = 0;
+
+    /** Is ZooKeeper Ready. */
+    private boolean zkReady;
+
+    /** Kafka Config. */
+    private KafkaConfig brokerConfig;
+
+    /** Kafka Server. */
+    private KafkaServer kafkaServer;
+
+    /** ZooKeeper Client. */
+    private ZkClient zkClient;
+
+    /** Embedded ZooKeeper. */
+    private EmbeddedZooKeeper zooKeeper;
+
+    /**
+     * Creates an embedded Kafka Broker.
+     */
+    public KafkaEmbeddedBroker() {
+        try {
+            setupEmbeddedZooKeeper();
+            setupEmbeddedKafkaServer();
+        }
+        catch (IOException | InterruptedException e) {
+            throw new RuntimeException("failed to start Kafka Broker " + e);
+        }
+
+    }
+
+    /**
+     * @return ZooKeeper Address.
+     */
+    public static String getZKAddress() {
+        return ZK_HOST + ":" + zkPort;
+    }
+
+    /**
+     * Creates a Topic.
+     *
+     * @param topic topic name
+     * @param partitions number of paritions for the topic
+     * @param replicationFactor replication factor
+     * @throws TimeoutException
+     * @throws InterruptedException
+     */
+    public void createTopic(String topic, final int partitions, final int replicationFactor)
+        throws TimeoutException, InterruptedException {
+        AdminUtils.createTopic(zkClient, topic, partitions, replicationFactor, new Properties());
+        waitUntilMetadataIsPropagated(topic, 0, 10000, 100);
+    }
+
+    /**
+     * Sends message to Kafka Broker.
+     *
+     * @param keyedMessages List of Keyed Messages.
+     * @return Producer used to send the message.
+     */
+    public Producer sendMessages(List<KeyedMessage<String, String>> keyedMessages)
{
+        Producer<String, String> producer = new Producer<>(getProducerConfig());
+        producer.send(scala.collection.JavaConversions.asScalaBuffer(keyedMessages));
+        return producer;
+    }
+
+    /**
+     * Shuts down Kafka Broker.
+     *
+     * @throws IOException
+     */
+    public void shutdown()
+        throws IOException {
+
+        zkReady = false;
+
+        if (kafkaServer != null)
+            kafkaServer.shutdown();
+
+        List<String> logDirs = scala.collection.JavaConversions.asJavaList(brokerConfig.logDirs());
+
+        for (String logDir : logDirs) {
+            FileUtils.deleteDirectory(new File(logDir));
+        }
+
+        if (zkClient != null) {
+            zkClient.close();
+            zkClient = null;
+        }
+
+        if (zooKeeper != null) {
+
+            try {
+                zooKeeper.shutdown();
+            }
+            catch (IOException e) {
+                // ignore
+            }
+
+            zooKeeper = null;
+        }
+
+    }
+
+    /**
+     * @return the Zookeeper Client
+     */
+    private ZkClient getZkClient() {
+        A.ensure(zkReady, "Zookeeper not setup yet");
+        A.notNull(zkClient, "Zookeeper client is not yet initialized");
+
+        return zkClient;
+    }
+
+    /**
+     * Checks if topic metadata is propagated.
+     *
+     * @param topic topic name
+     * @param partition partition
+     * @return true if propagated otherwise false
+     */
+    private boolean isMetadataPropagated(final String topic, final int partition) {
+        final scala.Option<PartitionStateInfo> partitionStateOption = kafkaServer.apis().metadataCache().getPartitionInfo(
+            topic, partition);
+        if (partitionStateOption.isDefined()) {
+            final PartitionStateInfo partitionState = partitionStateOption.get();
+            final LeaderAndIsr leaderAndInSyncReplicas = partitionState.leaderIsrAndControllerEpoch().leaderAndIsr();
+
+            if (ZkUtils.getLeaderForPartition(getZkClient(), topic, partition) != null
+                && Request.isValidBrokerId(leaderAndInSyncReplicas.leader())
+                && leaderAndInSyncReplicas.isr().size() >= 1)
+                return true;
+
+        }
+        return false;
+    }
+
+    /**
+     * Waits until metadata is propagated.
+     *
+     * @param topic topic name
+     * @param partition partition
+     * @param timeout timeout value in millis
+     * @param interval interval in millis to sleep
+     * @throws TimeoutException
+     * @throws InterruptedException
+     */
+    private void waitUntilMetadataIsPropagated(final String topic, final int partition, final
long timeout,
+        final long interval) throws TimeoutException, InterruptedException {
+        int attempt = 1;
+        final long startTime = System.currentTimeMillis();
+
+        while (true) {
+            if (isMetadataPropagated(topic, partition))
+                return;
+
+            final long duration = System.currentTimeMillis() - startTime;
+
+            if (duration < timeout)
+                Thread.sleep(interval);
+            else
+                throw new TimeoutException("metadata propagate timed out, attempt=" + attempt);
+
+            attempt++;
+        }
+
+    }
+
+    /**
+     * Sets up embedded Kafka Server
+     *
+     * @throws IOException
+     */
+    private void setupEmbeddedKafkaServer()
+        throws IOException {
+        A.ensure(zkReady, "Zookeeper should be setup before hand");
+
+        brokerConfig = new KafkaConfig(getBrokerConfig());
+        kafkaServer = new KafkaServer(brokerConfig, SystemTime$.MODULE$);
+        kafkaServer.startup();
+    }
+
+    /**
+     * Sets up embedded zooKeeper
+     *
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    private void setupEmbeddedZooKeeper()
+        throws IOException, InterruptedException {
+        EmbeddedZooKeeper zooKeeper = new EmbeddedZooKeeper(ZK_HOST, zkPort);
+        zooKeeper.startup();
+        zkPort = zooKeeper.getActualPort();
+        zkClient = new ZkClient(getZKAddress(), ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT,
ZKStringSerializer$.MODULE$);
+        zkReady = true;
+    }
+
+    /**
+     * @return Kafka Broker Address.
+     */
+    private static String getBrokerAddress() {
+        return ZK_HOST + ":" + BROKER_PORT;
+    }
+
+    /**
+     * Gets KafKa Brofer Config
+     *
+     * @return Kafka Broker Config
+     * @throws IOException
+     */
+    private static Properties getBrokerConfig()
+        throws IOException {
+        Properties props = new Properties();
+        props.put("broker.id", "0");
+        props.put("host.name", ZK_HOST);
+        props.put("port", "" + BROKER_PORT);
+        props.put("log.dir", createTempDir("_cfg").getAbsolutePath());
+        props.put("zookeeper.connect", getZKAddress());
+        props.put("log.flush.interval.messages", "1");
+        props.put("replica.socket.timeout.ms", "1500");
+        return props;
+    }
+
+    /**
+     * @return Kafka Producer Config
+     */
+    private static ProducerConfig getProducerConfig() {
+        Properties props = new Properties();
+        props.put("metadata.broker.list", getBrokerAddress());
+        props.put("serializer.class", "kafka.serializer.StringEncoder");
+        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
+        props.put("partitioner.class", "org.apache.ignite.kafka.SimplePartitioner");
+        return new ProducerConfig(props);
+    }
+
+    /**
+     * Creates Temp Directory
+     *
+     * @param prefix prefix
+     * @return Created File.
+     * @throws IOException
+     */
+    private static File createTempDir(final String prefix)
+        throws IOException {
+        final Path path = Files.createTempDirectory(prefix);
+        return path.toFile();
+
+    }
+
+    /**
+     * Creates Embedded ZooKeeper.
+     */
+    private static class EmbeddedZooKeeper {
+        /** Default ZooKeeper Host. */
+        private final String zkHost;
+
+        /** Default ZooKeeper Port. */
+        private final int zkPort;
+
+        /** NIO Context Factory. */
+        private NIOServerCnxnFactory factory;
+
+        /** Snapshot Directory. */
+        private File snapshotDir;
+
+        /** Log Directory. */
+        private File logDir;
+
+        /**
+         * Creates an embedded Zookeeper
+         * @param zkHost zookeeper host
+         * @param zkPort zookeeper port
+         */
+        EmbeddedZooKeeper(final String zkHost, final int zkPort) {
+            this.zkHost = zkHost;
+            this.zkPort = zkPort;
+        }
+
+        /**
+         * Starts up ZooKeeper.
+         *
+         * @throws IOException
+         * @throws InterruptedException
+         */
+        void startup()
+            throws IOException, InterruptedException {
+            snapshotDir = createTempDir("_ss");
+            logDir = createTempDir("_log");
+            ZooKeeperServer zooServer = new ZooKeeperServer(snapshotDir, logDir, 500);
+            factory = new NIOServerCnxnFactory();
+            factory.configure(new InetSocketAddress(zkHost, zkPort), 16);
+            factory.startup(zooServer);
+        }
+
+        /**
+         *
+         * @return actual port zookeeper is started
+         */
+        int getActualPort() {
+            return factory.getLocalPort();
+        }
+
+        /**
+         * Shuts down ZooKeeper.
+         *
+         * @throws IOException
+         */
+        void shutdown()
+            throws IOException {
+            if (factory != null) {
+                factory.shutdown();
+
+                U.delete(snapshotDir);
+                U.delete(logDir);
+            }
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2c41739d/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
new file mode 100644
index 0000000..5972639
--- /dev/null
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.ignite.stream.kafka;
+
+import org.apache.ignite.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import kafka.consumer.*;
+import kafka.producer.*;
+import kafka.serializer.*;
+import kafka.utils.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ * Tests {@link KafkaStreamer}.
+ */
+public class KafkaIgniteStreamerSelfTest
+    extends GridCommonAbstractTest {
+    /** Embedded Kafka. */
+    private KafkaEmbeddedBroker embeddedBroker;
+
+    /** Count. */
+    private static final int CNT = 100;
+
+    /** Test Topic. */
+    private static final String TOPIC_NAME = "page_visits";
+
+    /** Kafka Partition. */
+    private static final int PARTITIONS = 4;
+
+    /** Kafka Replication Factor. */
+    private static final int REPLICATION_FACTOR = 1;
+
+    /** Topic Message Key Prefix. */
+    private static final String KEY_PREFIX = "192.168.2.";
+
+    /** Topic Message Value Url. */
+    private static final String VALUE_URL = ",www.example.com,";
+
+    /** Constructor. */
+    public KafkaIgniteStreamerSelfTest() {
+        super(true);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    protected void beforeTest()
+        throws Exception {
+        grid().<Integer, String>getOrCreateCache(defaultCacheConfiguration());
+
+        embeddedBroker = new KafkaEmbeddedBroker();
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    protected void afterTest()
+        throws Exception {
+        grid().cache(null).clear();
+
+        embeddedBroker.shutdown();
+    }
+
+    /**
+     * Tests Kafka streamer.
+     *
+     * @throws TimeoutException
+     * @throws InterruptedException
+     */
+    public void testKafkaStreamer()
+        throws TimeoutException, InterruptedException {
+        embeddedBroker.createTopic(TOPIC_NAME, PARTITIONS, REPLICATION_FACTOR);
+
+        Map<String, String> keyValueMap = produceStream(TOPIC_NAME);
+        consumerStream(TOPIC_NAME, keyValueMap);
+    }
+
+    /**
+     * Produces/Sends messages to Kafka.
+     *
+     * @param topic Topic name.
+     * @return Map of key value messages.
+     */
+    private Map<String, String> produceStream(final String topic) {
+        final Map<String, String> keyValueMap = new HashMap<>();
+
+        // Generate random subnets.
+        List<Integer> subnet = new ArrayList<>();
+
+        int i = 0;
+        while (i <= CNT)
+            subnet.add(++i);
+
+        Collections.shuffle(subnet);
+
+        final List<KeyedMessage<String, String>> messages = new ArrayList<>();
+        for (int event = 0; event < CNT; event++) {
+            long runtime = new Date().getTime();
+            String ip = KEY_PREFIX + subnet.get(event);
+            String msg = runtime + VALUE_URL + ip;
+            messages.add(new KeyedMessage<>(topic, ip, msg));
+            keyValueMap.put(ip, msg);
+        }
+
+        final Producer<String, String> producer = embeddedBroker.sendMessages(messages);
+        producer.close();
+
+        return keyValueMap;
+    }
+
+    /**
+     * Consumes Kafka Stream via ignite.
+     *
+     * @param topic Topic name.
+     * @param keyValueMap Expected key value map.
+     * @throws TimeoutException TimeoutException.
+     * @throws InterruptedException InterruptedException.
+     */
+    private void consumerStream(final String topic, final Map<String, String> keyValueMap)
+        throws TimeoutException, InterruptedException {
+
+        KafkaStreamer<String, String, String> kafkaStmr = null;
+
+        final Ignite ignite = grid();
+        try (IgniteDataStreamer<String, String> stmr = ignite.dataStreamer(null)) {
+
+            stmr.allowOverwrite(true);
+            stmr.autoFlushFrequency(10);
+
+            // Configure socket streamer.
+            kafkaStmr = new KafkaStreamer<>();
+
+            // Get the cache.
+            IgniteCache<String, String> cache = ignite.cache(null);
+
+            // Set ignite instance.
+            kafkaStmr.setIgnite(ignite);
+
+            // Set data streamer instance.
+            kafkaStmr.setStreamer(stmr);
+
+            // Set the topic.
+            kafkaStmr.setTopic(topic);
+
+            // Set the number of threads.
+            kafkaStmr.setThreads(4);
+
+            // Set the consumer configuration.
+            kafkaStmr.setConsumerConfig(createDefaultConsumerConfig(KafkaEmbeddedBroker.getZKAddress(),
+                "groupX"));
+
+            // Set the decoders.
+            StringDecoder stringDecoder = new StringDecoder(new VerifiableProperties());
+            kafkaStmr.setKeyDecoder(stringDecoder);
+            kafkaStmr.setValueDecoder(stringDecoder);
+
+            // Start kafka streamer.
+            kafkaStmr.start();
+
+            final CountDownLatch latch = new CountDownLatch(CNT);
+            IgniteBiPredicate<UUID, CacheEvent> locLsnr = new IgniteBiPredicate<UUID,
CacheEvent>() {
+                @Override
+                public boolean apply(UUID uuid, CacheEvent evt) {
+                    latch.countDown();
+                    return true;
+                }
+            };
+
+            ignite.events(ignite.cluster().forCacheNodes(null)).remoteListen(locLsnr, null,
EVT_CACHE_OBJECT_PUT);
+            latch.await();
+
+            for (Map.Entry<String, String> entry : keyValueMap.entrySet()) {
+                final String key = entry.getKey();
+                final String value = entry.getValue();
+
+                final String cacheValue = cache.get(key);
+                assertEquals(value, cacheValue);
+            }
+        }
+
+        finally {
+            // Shutdown kafka streamer.
+            kafkaStmr.stop();
+        }
+    }
+
+    /**
+     * Creates default consumer config.
+     *
+     * @param zooKeeper Zookeeper address <server:port>.
+     * @param groupId Group Id for kafka subscriber.
+     * @return {@link ConsumerConfig} kafka consumer configuration.
+     */
+    private ConsumerConfig createDefaultConsumerConfig(String zooKeeper, String groupId)
{
+        A.notNull(zooKeeper, "zookeeper");
+        A.notNull(groupId, "groupId");
+
+        Properties props = new Properties();
+        props.put("zookeeper.connect", zooKeeper);
+        props.put("group.id", groupId);
+        props.put("zookeeper.session.timeout.ms", "400");
+        props.put("zookeeper.sync.time.ms", "200");
+        props.put("auto.commit.interval.ms", "1000");
+        props.put("auto.offset.reset", "smallest");
+
+        return new ConsumerConfig(props);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2c41739d/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
new file mode 100644
index 0000000..b836b44
--- /dev/null
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
@@ -0,0 +1,46 @@
+/*
+ * 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.ignite.stream.kafka;
+
+import kafka.producer.*;
+
+/**
+ * Simple Partitioner for Kafka.
+ */
+@SuppressWarnings("UnusedDeclaration")
+public class SimplePartitioner
+    implements Partitioner {
+
+    /**
+     * Partitions the key based on the key value.
+     *
+     * @param key Key.
+     * @param partitionSize Partition size.
+     * @return partition Partition.
+     */
+    public int partition(Object key, int partitionSize) {
+        int partition = 0;
+        String keyStr = (String)key;
+        String[] keyValues = keyStr.split("\\.");
+        Integer intKey = Integer.parseInt(keyValues[3]);
+        if (intKey > 0) {
+            partition = intKey % partitionSize;
+        }
+        return partition;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2c41739d/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index a6d1609..b47d34b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -70,6 +70,7 @@
         <module>modules/gce</module>
         <module>modules/cloud</module>
         <module>modules/mesos</module>
+        <module>modules/kafka</module>
     </modules>
 
     <profiles>


Mime
View raw message