diff --git a/modules/kafka-ext/README.txt b/modules/kafka-ext/README.txt
new file mode 100644
index 0000000..348d143
--- /dev/null
+++ b/modules/kafka-ext/README.txt
@@ -0,0 +1,210 @@
+Apache Ignite Kafka Streamer Module
+-----------------------------------
+
+Apache Ignite Kafka Streamer module provides streaming from Kafka to Ignite cache.
+
+There are two ways this can be achieved:
+- importing Kafka Streamer module in your Maven project and instantiate KafkaStreamer for data streaming;
+- using Kafka Connect functionality.
+
+Below are the details.
+
+## Importing Ignite Kafka Streamer Module In Maven Project
+
+If you are using Maven to manage dependencies of your project, you can add Kafka module
+dependency like this (replace '${ignite.version}' with actual Ignite version you are
+interested in):
+
+<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">
+    ...
+    <dependencies>
+        ...
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-kafka-ext</artifactId>
+            <version>${ignite.version}</version>
+        </dependency>
+        ...
+    </dependencies>
+    ...
+</project>
+
+
+## Streaming Data to Ignite via Kafka Connect
+
+Sink Connector will help you export data from Kafka to Ignite cache. It polls data from Kafka topics and writes it to the user-specified cache.
+For more information on Kafka Connect, see [Kafka Documentation](http://kafka.apache.org/documentation.html#connect).
+
+Connector can be found in 'optional/ignite-kafka.' It and its dependencies have to be on the classpath of a Kafka running instance,
+as described in the following subsection.
+
+### Setting up and Running
+
+1. Put the following jar files on Kafka's classpath
+- ignite-kafka-connect-x.x.x-SNAPSHOT.jar
+- ignite-core-x.x.x-SNAPSHOT.jar
+- ignite-spring-x.x.x-SNAPSHOT.jar
+- cache-api-1.0.0.jar
+- spring-aop-4.1.0.RELEASE.jar
+- spring-beans-4.1.0.RELEASE.jar
+- spring-context-4.1.0.RELEASE.jar
+- spring-core-4.1.0.RELEASE.jar
+- spring-expression-4.1.0.RELEASE.jar
+- commons-logging-1.1.1.jar
+
+2. Prepare worker configurations, e.g.,
+```
+bootstrap.servers=localhost:9092
+
+key.converter=org.apache.kafka.connect.storage.StringConverter
+value.converter=org.apache.kafka.connect.storage.StringConverter
+key.converter.schemas.enable=false
+value.converter.schemas.enable=false
+
+internal.key.converter=org.apache.kafka.connect.storage.StringConverter
+internal.value.converter=org.apache.kafka.connect.storage.StringConverter
+internal.key.converter.schemas.enable=false
+internal.value.converter.schemas.enable=false
+
+offset.storage.file.filename=/tmp/connect.offsets
+offset.flush.interval.ms=10000
+```
+
+3. Prepare connector configurations, e.g.,
+```
+# connector
+name=string-ignite-connector
+connector.class=org.apache.ignite.stream.kafka.connect.IgniteSinkConnector
+tasks.max=2
+topics=testTopic1,testTopic2
+
+# converter (optional)
+singleTupleExtractorCls=my.company.MyTupleExtractor
+
+# cache
+cacheName=cache1
+cacheAllowOverwrite=true
+igniteCfg=/some-path/ignite.xml
+```
+where 'cacheName' is the name of the cache you specify in '/some-path/ignite.xml' and the data from 'testTopic1,testTopic2'
+will be pulled and stored. 'cacheAllowOverwrite' is set to true if you want to enable overwriting existing values in cache.
+You can also set 'cachePerNodeDataSize' and 'cachePerNodeParOps' to adjust per-node buffer and the maximum number
+of parallel stream operations for a single node.
+If you need to create an Ignite key from a Kafka value, implement StreamSingleTupleExtractor and specify it as 'singleTupleExtractorCls'.
+
+See example-ignite.xml in tests for a simple cache configuration file example.
+
+4. Start connector, for instance, as follows,
+```
+./bin/connect-standalone.sh myconfig/connect-standalone.properties myconfig/ignite-connector.properties
+```
+
+## Checking the Flow
+
+To perform a very basic functionality check, you can do the following,
+
+1. Start Zookeeper
+```
+bin/zookeeper-server-start.sh config/zookeeper.properties
+```
+
+2. Start Kafka server
+```
+bin/kafka-server-start.sh config/server.properties
+```
+
+3. Provide some data input to the Kafka server
+```
+bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test --property parse.key=true --property key.separator=,
+k1,v1
+```
+
+4. Start the connector. For example,
+```
+./bin/connect-standalone.sh myconfig/connect-standalone.properties myconfig/ignite-connector.properties
+```
+
+5. Check the value is in the cache. For example, via REST,
+```
+http://node1:8080/ignite?cmd=size&cacheName=cache1
+```
+
+## Streaming Cache Event Data to Kafka via Kafka Connect
+
+Source connector enables listening to Ignite cache events and, upon filtering, stream them to Kafka.
+
+Connector can be found in 'optional/ignite-kafka.' It and its dependencies have to be on the classpath of a Kafka running instance,
+as described in the following subsection.
+
+### Setting up and Running
+
+1. Put the following jar files on Kafka's classpath
+- ignite-kafka-connect-x.x.x-SNAPSHOT.jar
+- ignite-core-x.x.x-SNAPSHOT.jar
+- cache-api-1.0.0.jar
+- ignite-spring-x.x.x-SNAPSHOT.jar
+- spring-aop-4.1.0.RELEASE.jar
+- spring-beans-4.1.0.RELEASE.jar
+- spring-context-4.1.0.RELEASE.jar
+- spring-core-4.1.0.RELEASE.jar
+- spring-expression-4.1.0.RELEASE.jar
+- commons-logging-1.1.1.jar
+
+2. Prepare worker configurations, e.g.,
+```
+bootstrap.servers=localhost:9092
+
+key.converter=org.apache.kafka.connect.storage.StringConverter
+value.converter=org.apache.ignite.stream.kafka.connect.serialization.CacheEventConverter
+key.converter.schemas.enable=false
+value.converter.schemas.enable=false
+
+internal.key.converter=org.apache.kafka.connect.storage.StringConverter
+internal.value.converter=org.apache.kafka.connect.storage.StringConverter
+internal.key.converter.schemas.enable=false
+internal.value.converter.schemas.enable=false
+
+offset.storage.file.filename=/tmp/connect.offsets
+offset.flush.interval.ms=10000
+```
+
+Note that the current implementation ignores key and schema of Kafka Connect, and stores marshalled cache events
+using org.apache.ignite.stream.kafka.connect.serialization.CacheEventConverter.
+
+3. Prepare connector configurations, e.g.,
+```
+# connector
+name=ignite-src-connector
+connector.class=org.apache.ignite.stream.kafka.connect.IgniteSourceConnector
+tasks.max=2
+
+# cache
+topicNames=testTopic1,testTopic2
+cacheEvts=put,removed
+## if you decide to filter remotely (recommended)
+cacheFilterCls=MyFilter
+cacheName=cache1
+igniteCfg=/some-path/ignite.xml
+```
+where 'cacheName' is the name of the cache you specify in '/some-path/ignite.xml' and the data from 'testTopic1,testTopic2'
+will be pulled and stored. Also consider using 'evtBufferSize' and 'evtBatchSize' for tuning the internal queue
+used to safely transfer data from Ignite cache to Kafka.
+
+The following cache events can be specified in the connector configurations:
+- CREATED
+- DESTROYED
+- PUT
+- READ
+- REMOVED
+- LOCKED
+- UNLOCKED
+- SWAPPED
+- UNSWAPPED
+- EXPIRED
+
+For a simple cache configuration file example, see example-ignite.xml in tests.
+
+4. Start the connector, as described in [Kafka Documentation](http://kafka.apache.org/documentation.html#connect).
diff --git a/modules/kafka-ext/licenses/apache-2.0.txt b/modules/kafka-ext/licenses/apache-2.0.txt
new file mode 100644
index 0000000..d645695
--- /dev/null
+++ b/modules/kafka-ext/licenses/apache-2.0.txt
@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed 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.
diff --git a/modules/kafka-ext/modules/core/src/test/config/log4j-test.xml b/modules/kafka-ext/modules/core/src/test/config/log4j-test.xml
new file mode 100755
index 0000000..b78fa9c
--- /dev/null
+++ b/modules/kafka-ext/modules/core/src/test/config/log4j-test.xml
@@ -0,0 +1,97 @@
+<?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.
+-->
+
+<!DOCTYPE log4j:configuration PUBLIC "-//APACHE//DTD LOG4J 1.2//EN"
+    "http://logging.apache.org/log4j/1.2/apidocs/org/apache/log4j/xml/doc-files/log4j.dtd">
+<!--
+    Log4j configuration.
+-->
+<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/" debug="false">
+    <!--
+        Logs System.out messages to console.
+    -->
+    <appender name="CONSOLE" class="org.apache.log4j.ConsoleAppender">
+        <!-- Log to STDOUT. -->
+        <param name="Target" value="System.out"/>
+
+        <!-- Log from DEBUG and higher. -->
+        <param name="Threshold" value="DEBUG"/>
+
+        <!-- The default pattern: Date Priority [Category] Message\n -->
+        <layout class="org.apache.log4j.PatternLayout">
+            <param name="ConversionPattern" value="[%d{ISO8601}][%-5p][%t][%c{1}] %m%n"/>
+        </layout>
+
+        <!-- Do not log beyond INFO level. -->
+        <filter class="org.apache.log4j.varia.LevelRangeFilter">
+            <param name="levelMin" value="DEBUG"/>
+            <param name="levelMax" value="INFO"/>
+        </filter>
+    </appender>
+
+    <!--
+        Logs all System.err messages to console.
+    -->
+    <appender name="CONSOLE_ERR" class="org.apache.log4j.ConsoleAppender">
+        <!-- Log to STDERR. -->
+        <param name="Target" value="System.err"/>
+
+        <!-- Log from WARN and higher. -->
+        <param name="Threshold" value="WARN"/>
+
+        <!-- The default pattern: Date Priority [Category] Message\n -->
+        <layout class="org.apache.log4j.PatternLayout">
+            <param name="ConversionPattern" value="[%d{ISO8601}][%-5p][%t][%c{1}] %m%n"/>
+        </layout>
+    </appender>
+
+    <!--
+        Logs all output to specified file.
+    -->
+    <appender name="FILE" class="org.apache.log4j.RollingFileAppender">
+        <param name="Threshold" value="DEBUG"/>
+        <param name="File" value="${IGNITE_HOME}/work/log/ignite.log"/>
+        <param name="Append" value="true"/>
+        <param name="MaxFileSize" value="10MB"/>
+        <param name="MaxBackupIndex" value="10"/>
+        <layout class="org.apache.log4j.PatternLayout">
+            <param name="ConversionPattern" value="[%d{ISO8601}][%-5p][%t][%c{1}] %m%n"/>
+        </layout>
+    </appender>
+
+    <!-- Disable all open source debugging. -->
+    <category name="org">
+        <level value="INFO"/>
+    </category>
+
+    <category name="org.eclipse.jetty">
+        <level value="INFO"/>
+    </category>
+
+    <!-- Default settings. -->
+    <root>
+        <!-- Print at info by default. -->
+        <level value="INFO"/>
+
+        <!-- Append to file and console. -->
+        <appender-ref ref="FILE"/>
+        <appender-ref ref="CONSOLE"/>
+        <appender-ref ref="CONSOLE_ERR"/>
+    </root>
+</log4j:configuration>
diff --git a/modules/kafka-ext/modules/core/src/test/config/tests.properties b/modules/kafka-ext/modules/core/src/test/config/tests.properties
new file mode 100644
index 0000000..18b3606
--- /dev/null
+++ b/modules/kafka-ext/modules/core/src/test/config/tests.properties
@@ -0,0 +1,153 @@
+#
+# 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.
+#
+
+# Local address to bind to.
+local.ip=127.0.0.1
+
+# TCP communication port
+comm.tcp.port=30010
+
+# JBoss JNDI
+# JBoss context factory for JNDI connection establishing.
+jboss.jndi.context.factory=org.jnp.interfaces.NamingContextFactory
+# JBoss specific parameter for JNDI connection establishing.
+jboss.jndi.pkg.prefixes=org.jboss.naming:org.jnp.interfaces
+# URL of JBoss server for the 1st node.
+jboss.jndi.node1.provider.url=jnp://localhost:1199
+# URL of JBoss server for the 2nd node.
+jboss.jndi.node2.provider.url=jnp://localhost:1299
+# JBoss Discovery test max wait time.
+jboss.disco.test.wait=180000
+
+# Deployment configuration paths.
+# You will either need to override deploy.uri.dir or supply CLASSES_URI as system property.
+#
+# Path to keystore with private and public keys.
+deploy.uri.secure.keystore=@{IGNITE_HOME}/modules/tests/config/securedeploy/keystore
+# Temporary dir where deployment unit stored before deploy.
+deploy.uri.tmpdir=${java.io.tmpdir}/gg
+# Deployment dir for file scanner test with different types of GAR's.
+deploy.uri.file2.path=${java.io.tmpdir}/gg/verification/
+# URI string.
+deploy.uri.file2=file://freq=200@localhost/${java.io.tmpdir}/gg/verification/
+# File scanner URI for local file deployment.
+deploy.uri.file=file://localhost/@{IGNITE_HOME}/modules/extdata/uri/target/file/
+# FTP scanner URI for FTP deployment.
+deploy.uri.ftp=ftp://ftptest:iddqd@94.72.60.102:21/test/deployment
+# Classes scanner URI for classes deployment. Must be overridden for every user.
+deploy.uri.cls=${CLASSES_URI}
+# Http scanner URI for HTTP deployment.
+deploy.uri.http=http://fake.uri
+# Http scanner URI for secure SSL HTTPs deployment.
+deploy.uri.https=https://fake.uri
+# Directory with descriptors to construct GAR files.
+deploy.gar.descriptor.dir=modules/urideploy/src/test/java/org/apache/ignite/spi/deployment/uri/META-INF
+
+# Directory with a number of descriptors for the Ant gar task.
+ant.gar.descriptor.dir=modules/extdata/p2p/META-INF
+# Temporary directory for the Ant task resulting GAR file.
+ant.gar.tmpdir=${java.io.tmpdir}/gg
+# The same as p2p.uri.cls but without protocol
+ant.gar.srcdir=@{IGNITE_HOME}/modules/extdata/uri/target/classes/
+
+# Paths to use in URI deployment SPI tests
+urideployment.jar.uri=modules/extdata/uri/target/deploy/uri.jar
+urideployment.path.tmp=modules/extdata/uri/target/deploy_tmp/
+
+# GAR paths to use in URI deployment SPI tests
+ant.urideployment.gar.uri=file://freq=5000@localhost/EXTDATA/uri/target/deploy
+ant.urideployment.gar.file=modules/extdata/uri/target/deploy/uri.gar
+ant.urideployment.gar.libs-file=modules/extdata/uri/target/deploy2/uri-libs.gar
+ant.urideployment.gar.classes-file=modules/extdata/uri/target/deploy2/uri-classes.gar
+ant.urideployment.gar.path=modules/extdata/uri/target/deploy/
+
+# Classpath directory for GridP2PUserVersionChangeSelfTest
+ant.userversion.class.dir=@{IGNITE_HOME}/modules/tests/java/
+
+# Multicast discovery self test.
+discovery.mbeanserver.selftest.baseport=50000
+
+# TCP communication self test.
+comm.mbeanserver.selftest.baseport=50100
+
+# Kernel tests.
+grid.comm.selftest.sender.timeout=1000
+grid.comm.selftest.timeout=10000
+
+#P2P tests
+#Overwrite this property. It should point to P2P module compilation directory.
+p2p.uri.cls=file://localhost/@{IGNITE_HOME}/modules/extdata/p2p/target/classes/
+p2p.uri.cls.second=file://localhost/@{IGNITE_HOME}/modules/extdata/uri/target/classes/
+
+# AOP tests.
+# Connector port for RMI.
+connector.rmi.port=7657
+# Connector port for XFire Web Service.
+connector.ws.port=9090
+
+# Load test duration in minutes.
+load.test.duration=500
+load.test.threadnum=50
+load.test.nodenum=5
+
+# Loaders tests
+loader.self.test.config=modules/core/src/test/config/loaders/grid-cfg.xml
+loader.self.multipletest.config=modules/core/src/test/config/loaders/grid-cfg-2-grids.xml
+loader.self.test.jboss.config=modules/core/src/test/config/loaders/grid-cfg.xml
+
+# WebSphere jmx properties
+websphere.jmx.connector.host=localhost
+websphere.jmx.connector.port=8880
+websphere.jmx.connector.security=false
+websphere.jmx.username=
+websphere.jmx.pwd=
+
+# GlassFish jmx properties for GlassFish Loader
+glassfish.jmx.rmi.connector.port=8686
+glassfish.jmx.username=admin
+glassfish.jmx.password=adminadmin
+
+# Tomcat jmx properties for Servlet Loader
+tomcat.jmx.rmi.connector.port=1097
+
+# Marshaller for tests
+#marshaller.class=org.apache.ignite.marshaller.jdk.GridJdkMarshaller
+
+# EC2 configuration for tests
+#amazon.access.key=
+#amazon.secret.key=
+
+# SSH config.
+ssh.username=uname
+ssh.password=passwd
+
+# SSL tests keystore.
+ssl.keystore.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/server.jks
+ssl.keystore.password=123456
+
+# node01 signed with trust-one, node02 and node03 by trust-two, node02old is expired
+# trust-both contains both CAs
+ssl.keystore.node01.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/ca/node01.jks
+ssl.keystore.node02.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/ca/node02.jks
+ssl.keystore.node03.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/ca/node03.jks
+ssl.keystore.trustone.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/ca/trust-one.jks
+ssl.keystore.trusttwo.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/ca/trust-two.jks
+ssl.keystore.trustboth.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/ca/trust-both.jks
+ssl.keystore.node02old.path=@{IGNITE_HOME}/modules/clients/src/test/keystore/ca/node02old.jks
+
+# Hadoop home directory.
+hadoop.home=@{HADOOP_HOME}
diff --git a/modules/kafka-ext/pom.xml b/modules/kafka-ext/pom.xml
new file mode 100644
index 0000000..115d767
--- /dev/null
+++ b/modules/kafka-ext/pom.xml
@@ -0,0 +1,155 @@
+<?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-extensions-parent</artifactId>
+        <version>1</version>
+        <relativePath>../../parent</relativePath>
+    </parent>
+
+    <artifactId>ignite-kafka-ext</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <url>http://ignite.apache.org</url>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-core</artifactId>
+            <version>${ignite.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>${kafka.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-api</artifactId>
+            <version>${kafka.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>${kafka.version}</version>
+            <classifier>test</classifier>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.11</artifactId>
+            <version>${kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-runtime</artifactId>
+            <version>${kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <version>${zookeeper.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>${curator.version}</version>
+            <scope>test</scope>
+            <!-- https://github.com/confluentinc/kafka-connect-elasticsearch/issues/143 -->
+            <exclusions>
+                <exclusion>
+                    <artifactId>guava</artifactId>
+                    <groupId>com.google.guava</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-spring</artifactId>
+            <version>${ignite.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-log4j</artifactId>
+            <version>${ignite.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.easymock</groupId>
+            <artifactId>easymock</artifactId>
+            <version>${easymock.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-core</artifactId>
+            <version>${ignite.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-tools</artifactId>
+            <version>${ignite.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.11</artifactId>
+            <version>${kafka.version}</version>
+            <classifier>test</classifier>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <!-- Generate the OSGi MANIFEST.MF for this bundle. -->
+            <plugin>
+                <groupId>org.apache.felix</groupId>
+                <artifactId>maven-bundle-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
new file mode 100644
index 0000000..02dbedd
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
@@ -0,0 +1,205 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.stream.StreamAdapter;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+/**
+ * Server that subscribes to topic messages from Kafka broker and streams its to key-value pairs into
+ * {@link IgniteDataStreamer} instance.
+ * <p>
+ * Uses Kafka's High Level Consumer API to read messages from Kafka.
+ */
+public class KafkaStreamer<K, V> extends StreamAdapter<ConsumerRecord, K, V> {
+    /** Default polling timeout. */
+    private static final long DFLT_TIMEOUT = 100;
+
+    /** Logger. */
+    private IgniteLogger log;
+
+    /** Polling tasks executor. */
+    private ExecutorService executor;
+
+    /** Topics. */
+    private List<String> topics;
+
+    /** Number of threads. */
+    private int threads;
+
+    /** Kafka consumer config. */
+    private Properties consumerCfg;
+
+    /** Polling timeout. */
+    private long timeout = DFLT_TIMEOUT;
+
+    /** Kafka consumer tasks. */
+    private final List<ConsumerTask> consumerTasks = new ArrayList<>();
+
+    /**
+     * Sets the topic names.
+     *
+     * @param topics Topic names.
+     */
+    public void setTopic(List<String> topics) {
+        this.topics = topics;
+    }
+
+    /**
+     * Sets the threads.
+     *
+     * @param threads Number of threads.
+     */
+    public void setThreads(int threads) {
+        this.threads = threads;
+    }
+
+    /**
+     * Sets the consumer config.
+     *
+     * @param consumerCfg Consumer configuration.
+     */
+    public void setConsumerConfig(Properties consumerCfg) {
+        this.consumerCfg = consumerCfg;
+    }
+
+    /**
+     * Sets the polling timeout for Kafka tasks.
+     *
+     * @param timeout Timeout.
+     */
+    public void setTimeout(long timeout) {
+        A.ensure(timeout > 0, "timeout > 0");
+
+        this.timeout = timeout;
+    }
+
+    /**
+     * Starts streamer.
+     *
+     * @throws IgniteException If failed.
+     */
+    public void start() {
+        A.notNull(getStreamer(), "streamer");
+        A.notNull(getIgnite(), "ignite");
+        A.notNull(topics, "topics");
+        A.notNull(consumerCfg, "kafka consumer config");
+        A.ensure(threads > 0, "threads > 0");
+        A.ensure(null != getSingleTupleExtractor() || null != getMultipleTupleExtractor(),
+            "Extractor must be configured");
+
+        log = getIgnite().log();
+
+        executor = Executors.newFixedThreadPool(threads);
+
+        IntStream.range(0, threads).forEach(i -> consumerTasks.add(new ConsumerTask(consumerCfg)));
+
+        for (ConsumerTask task : consumerTasks)
+            executor.submit(task);
+    }
+
+    /**
+     * Stops streamer.
+     */
+    public void stop() {
+        for (ConsumerTask task : consumerTasks)
+            task.stop();
+
+        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 ignored) {
+                if (log.isDebugEnabled())
+                    log.debug("Interrupted during shutdown, exiting uncleanly.");
+            }
+        }
+    }
+
+    /** Polling task. */
+    class ConsumerTask implements Callable<Void> {
+        /** Kafka consumer. */
+        private final KafkaConsumer<?, ?> consumer;
+
+        /** Stopped. */
+        private volatile boolean stopped;
+
+        /** Constructor. */
+        public ConsumerTask(Properties consumerCfg) {
+            this.consumer = new KafkaConsumer<>(consumerCfg);
+        }
+
+        /** {@inheritDoc} */
+        @Override public Void call() {
+            consumer.subscribe(topics);
+
+            try {
+                while (!stopped) {
+                    for (ConsumerRecord record : consumer.poll(timeout)) {
+                        try {
+                            addMessage(record);
+                        }
+                        catch (Exception e) {
+                            U.error(log, "Record is ignored due to an error [record = " + record + ']', e);
+                        }
+                    }
+                }
+            }
+            catch (WakeupException we) {
+                if (log.isInfoEnabled())
+                    log.info("Consumer is being stopped.");
+            }
+            catch (KafkaException ke) {
+                log.error("Kafka error", ke);
+            }
+            finally {
+                consumer.close();
+            }
+
+            return null;
+        }
+
+        /** Stops the polling task. */
+        public void stop() {
+            stopped = true;
+
+            if (consumer != null)
+                consumer.wakeup();
+        }
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnector.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnector.java
new file mode 100644
index 0000000..3fbfd9c
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnector.java
@@ -0,0 +1,100 @@
+/*
+ * 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.connect;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.sink.SinkConnector;
+
+/**
+ * Sink connector to manage sink tasks that transfer Kafka data to Ignite grid.
+ */
+public class IgniteSinkConnector extends SinkConnector {
+    /** Sink properties. */
+    private Map<String, String> configProps;
+
+    /** Expected configurations. */
+    private static final ConfigDef CONFIG_DEF = new ConfigDef();
+
+    /** {@inheritDoc} */
+    @Override public String version() {
+        return AppInfoParser.getVersion();
+    }
+
+    /**
+     * A sink lifecycle method. Validates grid-specific sink properties.
+     *
+     * @param props Sink properties.
+     */
+    @Override public void start(Map<String, String> props) {
+        configProps = props;
+
+        try {
+            A.notNullOrEmpty(configProps.get(SinkConnector.TOPICS_CONFIG), "topics");
+            A.notNullOrEmpty(configProps.get(IgniteSinkConstants.CACHE_NAME), "cache name");
+            A.notNullOrEmpty(configProps.get(IgniteSinkConstants.CACHE_CFG_PATH), "path to cache config file");
+        }
+        catch (IllegalArgumentException e) {
+            throw new ConnectException("Cannot start IgniteSinkConnector due to configuration error", e);
+        }
+    }
+
+    /**
+     * Obtains a sink task class to be instantiated for feeding data into grid.
+     *
+     * @return IgniteSinkTask class.
+     */
+    @Override public Class<? extends Task> taskClass() {
+        return IgniteSinkTask.class;
+    }
+
+    /**
+     * Builds each config for <tt>maxTasks</tt> tasks.
+     *
+     * @param maxTasks Max number of tasks.
+     * @return Task configs.
+     */
+    @Override public List<Map<String, String>> taskConfigs(int maxTasks) {
+        List<Map<String, String>> taskConfigs = new ArrayList<>();
+        Map<String, String> taskProps = new HashMap<>();
+
+        taskProps.putAll(configProps);
+
+        for (int i = 0; i < maxTasks; i++)
+            taskConfigs.add(taskProps);
+
+        return taskConfigs;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void stop() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public ConfigDef config() {
+        return CONFIG_DEF;
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConstants.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConstants.java
new file mode 100644
index 0000000..3fb511a
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConstants.java
@@ -0,0 +1,41 @@
+/*
+ * 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.connect;
+
+/**
+ * Sink configuration strings.
+ */
+public class IgniteSinkConstants {
+    /** Ignite configuration file path. */
+    public static final String CACHE_CFG_PATH = "igniteCfg";
+
+    /** Cache name. */
+    public static final String CACHE_NAME = "cacheName";
+
+    /** Flag to enable overwriting existing values in cache. */
+    public static final String CACHE_ALLOW_OVERWRITE = "cacheAllowOverwrite";
+
+    /** Size of per-node buffer before data is sent to remote node. */
+    public static final String CACHE_PER_NODE_DATA_SIZE = "cachePerNodeDataSize";
+
+    /** Maximum number of parallel stream operations per node. */
+    public static final String CACHE_PER_NODE_PAR_OPS = "cachePerNodeParOps";
+
+    /** Class to transform the entry before feeding into cache. */
+    public static final String SINGLE_TUPLE_EXTRACTOR_CLASS = "singleTupleExtractorCls";
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkTask.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkTask.java
new file mode 100644
index 0000000..184f309
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSinkTask.java
@@ -0,0 +1,201 @@
+/*
+ * 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.connect;
+
+import java.util.Collection;
+import java.util.Map;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.stream.StreamSingleTupleExtractor;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.apache.kafka.connect.sink.SinkTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Task to consume sequences of SinkRecords and write data to grid.
+ */
+public class IgniteSinkTask extends SinkTask {
+    /** Logger. */
+    private static final Logger log = LoggerFactory.getLogger(IgniteSinkTask.class);
+
+    /** Flag for stopped state. */
+    private static volatile boolean stopped = true;
+
+    /** Ignite grid configuration file. */
+    private static String igniteConfigFile;
+
+    /** Cache name. */
+    private static String cacheName;
+
+    /** Entry transformer. */
+    private static StreamSingleTupleExtractor<SinkRecord, Object, Object> extractor;
+
+    /** {@inheritDoc} */
+    @Override public String version() {
+        return new IgniteSinkConnector().version();
+    }
+
+    /**
+     * Initializes grid client from configPath.
+     *
+     * @param props Task properties.
+     */
+    @Override public void start(Map<String, String> props) {
+        // Each task has the same parameters -- avoid setting more than once.
+        if (cacheName != null)
+            return;
+
+        cacheName = props.get(IgniteSinkConstants.CACHE_NAME);
+        igniteConfigFile = props.get(IgniteSinkConstants.CACHE_CFG_PATH);
+
+        if (props.containsKey(IgniteSinkConstants.CACHE_ALLOW_OVERWRITE))
+            StreamerContext.getStreamer().allowOverwrite(
+                Boolean.parseBoolean(props.get(IgniteSinkConstants.CACHE_ALLOW_OVERWRITE)));
+
+        if (props.containsKey(IgniteSinkConstants.CACHE_PER_NODE_DATA_SIZE))
+            StreamerContext.getStreamer().perNodeBufferSize(
+                Integer.parseInt(props.get(IgniteSinkConstants.CACHE_PER_NODE_DATA_SIZE)));
+
+        if (props.containsKey(IgniteSinkConstants.CACHE_PER_NODE_PAR_OPS))
+            StreamerContext.getStreamer().perNodeParallelOperations(
+                Integer.parseInt(props.get(IgniteSinkConstants.CACHE_PER_NODE_PAR_OPS)));
+
+        if (props.containsKey(IgniteSinkConstants.SINGLE_TUPLE_EXTRACTOR_CLASS)) {
+            String transformerCls = props.get(IgniteSinkConstants.SINGLE_TUPLE_EXTRACTOR_CLASS);
+            if (transformerCls != null && !transformerCls.isEmpty()) {
+                try {
+                    Class<? extends StreamSingleTupleExtractor> clazz =
+                        (Class<? extends StreamSingleTupleExtractor<SinkRecord, Object, Object>>)
+                            Class.forName(transformerCls);
+
+                    extractor = clazz.newInstance();
+                }
+                catch (Exception e) {
+                    throw new ConnectException("Failed to instantiate the provided transformer!", e);
+                }
+            }
+        }
+
+        stopped = false;
+    }
+
+    /**
+     * Buffers records.
+     *
+     * @param records Records to inject into grid.
+     */
+    @SuppressWarnings("unchecked")
+    @Override public void put(Collection<SinkRecord> records) {
+        try {
+            for (SinkRecord record : records) {
+                // Data is flushed asynchronously when CACHE_PER_NODE_DATA_SIZE is reached.
+                if (extractor != null) {
+                    Map.Entry<Object, Object> entry = extractor.extract(record);
+                    StreamerContext.getStreamer().addData(entry.getKey(), entry.getValue());
+                }
+                else {
+                    if (record.key() != null) {
+                        StreamerContext.getStreamer().addData(record.key(), record.value());
+                    }
+                    else {
+                        log.error("Failed to stream a record with null key!");
+                    }
+                }
+            }
+        }
+        catch (ConnectException e) {
+            log.error("Failed adding record", e);
+
+            throw new ConnectException(e);
+        }
+    }
+
+    /**
+     * Pushes buffered data to grid. Flush interval is configured by worker configurations.
+     *
+     * @param offsets Offset information.
+     */
+    @Override public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
+        if (stopped)
+            return;
+
+        StreamerContext.getStreamer().flush();
+    }
+
+    /**
+     * Stops the grid client.
+     */
+    @Override public void stop() {
+        if (stopped)
+            return;
+
+        stopped = true;
+
+        StreamerContext.getIgnite().close();
+    }
+
+    /**
+     * Used by unit test to avoid restart node and valid state of the <code>stopped</code> flag.
+     *
+     * @param stopped Stopped flag.
+     */
+    protected static void setStopped(boolean stopped) {
+        IgniteSinkTask.stopped = stopped;
+
+        extractor = null;
+    }
+
+    /**
+     * Streamer context initializing grid and data streamer instances on demand.
+     */
+    public static class StreamerContext {
+        /** Constructor. */
+        private StreamerContext() {
+        }
+
+        /** Instance holder. */
+        private static class Holder {
+            private static final Ignite IGNITE = Ignition.start(igniteConfigFile);
+
+            private static final IgniteDataStreamer STREAMER = IGNITE.dataStreamer(cacheName);
+        }
+
+        /**
+         * Obtains grid instance.
+         *
+         * @return Grid instance.
+         */
+        public static Ignite getIgnite() {
+            return Holder.IGNITE;
+        }
+
+        /**
+         * Obtains data streamer instance.
+         *
+         * @return Data streamer instance.
+         */
+        public static IgniteDataStreamer getStreamer() {
+            return Holder.STREAMER;
+        }
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnector.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnector.java
new file mode 100644
index 0000000..986888e
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnector.java
@@ -0,0 +1,90 @@
+/*
+ * 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.connect;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.source.SourceConnector;
+
+/**
+ * Source connector to manage source tasks that listens to registered Ignite grid events and forward them to Kafka.
+ *
+ * Note that only cache events are enabled for streaming.
+ */
+public class IgniteSourceConnector extends SourceConnector {
+    /** Source properties. */
+    private Map<String, String> configProps;
+
+    /** Expected configurations. */
+    private static final ConfigDef CONFIG_DEF = new ConfigDef();
+
+    /** {@inheritDoc} */
+    @Override public String version() {
+        return AppInfoParser.getVersion();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void start(Map<String, String> props) {
+        try {
+            A.notNullOrEmpty(props.get(IgniteSourceConstants.CACHE_NAME), "cache name");
+            A.notNullOrEmpty(props.get(IgniteSourceConstants.CACHE_CFG_PATH), "path to cache config file");
+            A.notNullOrEmpty(props.get(IgniteSourceConstants.CACHE_EVENTS), "Registered cache events");
+            A.notNullOrEmpty(props.get(IgniteSourceConstants.TOPIC_NAMES), "Kafka topics");
+        }
+        catch (IllegalArgumentException e) {
+            throw new ConnectException("Cannot start IgniteSourceConnector due to configuration error", e);
+        }
+
+        configProps = props;
+    }
+
+    /** {@inheritDoc} */
+    @Override public Class<? extends Task> taskClass() {
+        return IgniteSourceTask.class;
+    }
+
+    /** {@inheritDoc} */
+    @Override public List<Map<String, String>> taskConfigs(int maxTasks) {
+        List<Map<String, String>> taskConfigs = new ArrayList<>();
+        Map<String, String> taskProps = new HashMap<>();
+
+        taskProps.putAll(configProps);
+
+        for (int i = 0; i < maxTasks; i++)
+            taskConfigs.add(taskProps);
+
+        return taskConfigs;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void stop() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public ConfigDef config() {
+        return CONFIG_DEF;
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConstants.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConstants.java
new file mode 100644
index 0000000..7d590e5
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConstants.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.ignite.stream.kafka.connect;
+
+/**
+ * Sink configuration strings.
+ */
+public class IgniteSourceConstants {
+    /** Ignite configuration file path. */
+    public static final String CACHE_CFG_PATH = "igniteCfg";
+
+    /** Cache name. */
+    public static final String CACHE_NAME = "cacheName";
+
+    /** Events to be listened to. Names corresponds to {@link IgniteSourceTask.CacheEvt}. */
+    public static final String CACHE_EVENTS = "cacheEvts";
+
+    /** Internal buffer size. */
+    public static final String INTL_BUF_SIZE = "evtBufferSize";
+
+    /** Size of one chunk drained from the internal buffer. */
+    public static final String INTL_BATCH_SIZE = "evtBatchSize";
+
+    /** User-defined filter class. */
+    public static final String CACHE_FILTER_CLASS = "cacheFilterCls";
+
+    /** Kafka topic. */
+    public static final String TOPIC_NAMES = "topicNames";
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceTask.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceTask.java
new file mode 100644
index 0000000..6e3d43b
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceTask.java
@@ -0,0 +1,359 @@
+/*
+ * 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.connect;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.source.SourceTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Task to consume remote cluster cache events from the grid and inject them into Kafka.
+ * <p>
+ * Note that a task will create a bounded queue in the grid for more reliable data transfer.
+ * Queue size can be changed by {@link IgniteSourceConstants#INTL_BUF_SIZE}.
+ */
+public class IgniteSourceTask extends SourceTask {
+    /** Logger. */
+    private static final Logger log = LoggerFactory.getLogger(IgniteSourceTask.class);
+
+    /** Tasks static monitor. */
+    private static final Object lock = new Object();
+
+    /** Event buffer size. */
+    private static int evtBufSize = 100000;
+
+    /** Event buffer. */
+    private static BlockingQueue<CacheEvent> evtBuf = new LinkedBlockingQueue<>(evtBufSize);
+
+    /** Max number of events taken from the buffer at once. */
+    private static int evtBatchSize = 100;
+
+    /** Flag for stopped state. */
+    private static volatile boolean stopped = true;
+
+    /** Ignite grid configuration file. */
+    private static String igniteCfgFile;
+
+    /** Cache name. */
+    private static String cacheName;
+
+    /** Remote Listener id. */
+    private static UUID rmtLsnrId;
+
+    /** Local listener. */
+    private static TaskLocalListener locLsnr = new TaskLocalListener();
+
+    /** User-defined filter. */
+    private static IgnitePredicate<CacheEvent> filter;
+
+    /** Topic. */
+    private static String topics[];
+
+    /** Offset. */
+    private static final Map<String, Long> offset = Collections.singletonMap("offset", 0L);
+
+    /** Partition. */
+    private static final Map<String, String> srcPartition = Collections.singletonMap("cache", null);
+
+    /** {@inheritDoc} */
+    @Override public String version() {
+        return new IgniteSinkConnector().version();
+    }
+
+    /**
+     * Filtering is done remotely. Local listener buffers data for injection into Kafka.
+     *
+     * @param props Task properties.
+     */
+    @Override public void start(Map<String, String> props) {
+        synchronized (lock) {
+            // Each task has the same parameters -- avoid setting more than once.
+            // Nothing to do if the task has been already started.
+            if (!stopped)
+                return;
+
+            cacheName = props.get(IgniteSourceConstants.CACHE_NAME);
+            igniteCfgFile = props.get(IgniteSourceConstants.CACHE_CFG_PATH);
+            topics = props.get(IgniteSourceConstants.TOPIC_NAMES).split("\\s*,\\s*");
+
+            if (props.containsKey(IgniteSourceConstants.INTL_BUF_SIZE))
+                evtBufSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BUF_SIZE));
+
+            if (props.containsKey(IgniteSourceConstants.INTL_BATCH_SIZE))
+                evtBatchSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BATCH_SIZE));
+
+            if (props.containsKey(IgniteSourceConstants.CACHE_FILTER_CLASS)) {
+                String filterCls = props.get(IgniteSourceConstants.CACHE_FILTER_CLASS);
+                if (filterCls != null && !filterCls.isEmpty()) {
+                    try {
+                        Class<? extends IgnitePredicate<CacheEvent>> clazz =
+                            (Class<? extends IgnitePredicate<CacheEvent>>)Class.forName(filterCls);
+
+                        filter = clazz.newInstance();
+                    }
+                    catch (Exception e) {
+                        log.error("Failed to instantiate the provided filter! " +
+                            "User-enabled filtering is ignored!", e);
+                    }
+                }
+            }
+
+            TaskRemoteFilter rmtLsnr = new TaskRemoteFilter(cacheName);
+
+            try {
+                int[] evts = cacheEvents(props.get(IgniteSourceConstants.CACHE_EVENTS));
+
+                rmtLsnrId = IgniteGrid.getIgnite().events(IgniteGrid.getIgnite().cluster().forCacheNodes(cacheName))
+                    .remoteListen(locLsnr, rmtLsnr, evts);
+            }
+            catch (Exception e) {
+                log.error("Failed to register event listener!", e);
+
+                throw new ConnectException(e);
+            }
+            finally {
+                stopped = false;
+            }
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public List<SourceRecord> poll() throws InterruptedException {
+        ArrayList<SourceRecord> records = new ArrayList<>(evtBatchSize);
+        ArrayList<CacheEvent> evts = new ArrayList<>(evtBatchSize);
+
+        if (stopped)
+            return records;
+
+        try {
+            if (evtBuf.drainTo(evts, evtBatchSize) > 0) {
+                for (CacheEvent evt : evts) {
+                    // schema and keys are ignored.
+                    for (String topic : topics)
+                        records.add(new SourceRecord(srcPartition, offset, topic, null, evt));
+                }
+
+                return records;
+            }
+        }
+        catch (IgniteException e) {
+            log.error("Error when polling event queue!", e);
+        }
+
+        // for shutdown.
+        return null;
+    }
+
+    /**
+     * Converts comma-delimited cache events strings to Ignite internal representation.
+     *
+     * @param evtPropsStr Comma-delimited cache event names.
+     * @return Ignite internal representation of cache events to be registered with the remote listener.
+     * @throws Exception If error.
+     */
+    private int[] cacheEvents(String evtPropsStr) throws Exception {
+        String[] evtStr = evtPropsStr.split("\\s*,\\s*");
+
+        if (evtStr.length == 0)
+            return EventType.EVTS_CACHE;
+
+        int[] evts = new int[evtStr.length];
+
+        try {
+            for (int i = 0; i < evtStr.length; i++)
+                evts[i] = CacheEvt.valueOf(evtStr[i].toUpperCase()).getId();
+        }
+        catch (Exception e) {
+            log.error("Failed to recognize the provided cache event!", e);
+
+            throw new Exception(e);
+        }
+        return evts;
+    }
+
+    /**
+     * Stops the grid client.
+     */
+    @Override public synchronized void stop() {
+        if (stopped)
+            return;
+
+        stopped = true;
+
+        stopRemoteListen();
+
+        IgniteGrid.getIgnite().close();
+    }
+
+    /**
+     * Stops the remote listener.
+     */
+    protected void stopRemoteListen() {
+        if (rmtLsnrId != null)
+            IgniteGrid.getIgnite().events(IgniteGrid.getIgnite().cluster().forCacheNodes(cacheName))
+                .stopRemoteListen(rmtLsnrId);
+
+        rmtLsnrId = null;
+    }
+
+    /**
+     * Used by unit test to avoid restart node and valid state of the <code>stopped</code> flag.
+     *
+     * @param stopped Stopped flag.
+     */
+    protected static void setStopped(boolean stopped) {
+        IgniteSourceTask.stopped = stopped;
+    }
+
+    /**
+     * Local listener buffering cache events to be further sent to Kafka.
+     */
+    private static class TaskLocalListener implements IgniteBiPredicate<UUID, CacheEvent> {
+        /** {@inheritDoc} */
+        @Override public boolean apply(UUID id, CacheEvent evt) {
+            try {
+                if (!evtBuf.offer(evt, 10, TimeUnit.MILLISECONDS))
+                    log.error("Failed to buffer event {}", evt.name());
+            }
+            catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+
+            return true;
+        }
+    }
+
+    /**
+     * Remote filter.
+     */
+    private static class TaskRemoteFilter implements IgnitePredicate<CacheEvent> {
+        /** */
+        @IgniteInstanceResource
+        Ignite ignite;
+
+        /** Cache name. */
+        private final String cacheName;
+
+        /**
+         * @param cacheName Cache name.
+         */
+        TaskRemoteFilter(String cacheName) {
+            this.cacheName = cacheName;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean apply(CacheEvent evt) {
+            Affinity<Object> affinity = ignite.affinity(cacheName);
+
+            if (affinity.isPrimary(ignite.cluster().localNode(), evt.key())) {
+                // Process this event. Ignored on backups.
+                if (filter != null && filter.apply(evt))
+                    return false;
+
+                return true;
+            }
+
+            return false;
+        }
+    }
+
+    /**
+     * Grid instance initialized on demand.
+     */
+    private static class IgniteGrid {
+        /** Constructor. */
+        private IgniteGrid() {
+            // No-op.
+        }
+
+        /** Instance holder. */
+        private static class Holder {
+            /** */
+            private static final Ignite IGNITE = Ignition.start(igniteCfgFile);
+        }
+
+        /**
+         * Obtains grid instance.
+         *
+         * @return Grid instance.
+         */
+        private static Ignite getIgnite() {
+            return Holder.IGNITE;
+        }
+    }
+
+    /** Cache events available for listening. */
+    private enum CacheEvt {
+        /** */
+        CREATED(EventType.EVT_CACHE_ENTRY_CREATED),
+        /** */
+        DESTROYED(EventType.EVT_CACHE_ENTRY_DESTROYED),
+        /** */
+        PUT(EventType.EVT_CACHE_OBJECT_PUT),
+        /** */
+        READ(EventType.EVT_CACHE_OBJECT_READ),
+        /** */
+        REMOVED(EventType.EVT_CACHE_OBJECT_REMOVED),
+        /** */
+        LOCKED(EventType.EVT_CACHE_OBJECT_LOCKED),
+        /** */
+        UNLOCKED(EventType.EVT_CACHE_OBJECT_UNLOCKED),
+        /** */
+        EXPIRED(EventType.EVT_CACHE_OBJECT_EXPIRED);
+
+        /** Internal Ignite event id. */
+        private final int id;
+
+        /**
+         * Constructor.
+         *
+         * @param id Internal Ignite event id.
+         */
+        CacheEvt(int id) {
+            this.id = id;
+        }
+
+        /**
+         * Gets Ignite event id.
+         *
+         * @return Ignite event id.
+         */
+        int getId() {
+            return id;
+        }
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/package-info.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/package-info.java
new file mode 100644
index 0000000..f565c04
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Contains implementation of Kafka Streamer Connector.
+ */
+
+package org.apache.ignite.stream.kafka.connect;
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventConverter.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventConverter.java
new file mode 100644
index 0000000..6a60bad
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventConverter.java
@@ -0,0 +1,67 @@
+/*
+ * 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.connect.serialization;
+
+import java.util.Map;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaAndValue;
+import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.storage.Converter;
+
+/**
+ * {@link CacheEvent} converter for Connect API.
+ */
+public class CacheEventConverter implements Converter {
+    private final CacheEventDeserializer deserializer = new CacheEventDeserializer();
+
+    private final CacheEventSerializer serializer = new CacheEventSerializer();
+
+    /** {@inheritDoc} */
+    @Override public void configure(Map<String, ?> map, boolean b) {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte[] fromConnectData(String topic, Schema schema, Object o) {
+        try {
+            return serializer.serialize(topic, (CacheEvent)o);
+        }
+        catch (SerializationException e) {
+            throw new DataException("Failed to convert to byte[] due to a serialization error", e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public SchemaAndValue toConnectData(String topic, byte[] bytes) {
+        CacheEvent evt;
+
+        try {
+            evt = deserializer.deserialize(topic, bytes);
+        }
+        catch (SerializationException e) {
+            throw new DataException("Failed to convert to Kafka Connect data due to a serialization error", e);
+        }
+
+        if (evt == null) {
+            return SchemaAndValue.NULL;
+        }
+        return new SchemaAndValue(null, evt);
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventDeserializer.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventDeserializer.java
new file mode 100644
index 0000000..3bcf5fb
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventDeserializer.java
@@ -0,0 +1,55 @@
+/*
+ * 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.connect.serialization;
+
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.marshaller.jdk.JdkMarshaller;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+/**
+ * Deserializer based on {@link JdkMarshaller}.
+ */
+public class CacheEventDeserializer implements Deserializer<CacheEvent> {
+    /** Marshaller. */
+    private static final Marshaller marsh = new JdkMarshaller();
+
+    /** {@inheritDoc} */
+    @Override public void configure(Map<String, ?> map, boolean b) {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public CacheEvent deserialize(String topic, byte[] bytes) {
+        try {
+            return U.unmarshal(marsh, bytes, getClass().getClassLoader());
+        }
+        catch (IgniteCheckedException e) {
+            throw new SerializationException("Failed to deserialize cache event!", e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void close() {
+        // No-op.
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventSerializer.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventSerializer.java
new file mode 100644
index 0000000..bc09256
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/CacheEventSerializer.java
@@ -0,0 +1,55 @@
+/*
+ * 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.connect.serialization;
+
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.marshaller.jdk.JdkMarshaller;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.serialization.Serializer;
+
+/**
+ * Serializer based on {@link JdkMarshaller}.
+ */
+public class CacheEventSerializer implements Serializer<CacheEvent> {
+    /** Marshaller. */
+    private static final Marshaller marsh = new JdkMarshaller();
+
+    /** {@inheritDoc} */
+    @Override public void configure(Map<String, ?> map, boolean b) {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte[] serialize(String topic, CacheEvent event) {
+        try {
+            return U.marshal(marsh, event);
+        }
+        catch (IgniteCheckedException e) {
+            throw new SerializationException("Failed to serialize cache event!", e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void close() {
+        // No-op.
+    }
+}
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/package-info.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/package-info.java
new file mode 100644
index 0000000..726ab39
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/serialization/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Contains implementation of Kafka Streamer Serializer.
+ */
+
+package org.apache.ignite.stream.kafka.connect.serialization;
diff --git a/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/package-info.java b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/package-info.java
new file mode 100644
index 0000000..d668d30
--- /dev/null
+++ b/modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/package-info.java
@@ -0,0 +1,22 @@
+/*
+ *  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.
+ */
+
+/**
+ * Contains implementation of Kafka Streamer.
+ */
+
+package org.apache.ignite.stream.kafka;
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/IgniteKafkaStreamerSelfTestSuite.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/IgniteKafkaStreamerSelfTestSuite.java
new file mode 100644
index 0000000..c88fba8
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/IgniteKafkaStreamerSelfTestSuite.java
@@ -0,0 +1,38 @@
+/*
+ * 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.stream.kafka.connect.IgniteSinkConnectorTest;
+import org.apache.ignite.stream.kafka.connect.IgniteSourceConnectorTest;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+/**
+ * Apache Kafka streamers tests.
+ */
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+    // Kafka streamer.
+    KafkaIgniteStreamerSelfTest.class,
+
+    // Kafka streamers via Connect API.
+    IgniteSinkConnectorTest.class,
+    IgniteSourceConnectorTest.class
+})
+public class IgniteKafkaStreamerSelfTestSuite {
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
new file mode 100644
index 0000000..3f20d87
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
@@ -0,0 +1,275 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.Test;
+
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
+
+/**
+ * Tests {@link KafkaStreamer}.
+ */
+public class KafkaIgniteStreamerSelfTest extends GridCommonAbstractTest {
+    /** Embedded Kafka. */
+    private TestKafkaBroker 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 IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        return super.getConfiguration(igniteInstanceName).setIncludeEventTypes(EventType.EVTS_ALL);
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("unchecked")
+    @Override protected void beforeTest() throws Exception {
+        grid().<Integer, String>getOrCreateCache(defaultCacheConfiguration());
+
+        embeddedBroker = new TestKafkaBroker();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        grid().cache(DEFAULT_CACHE_NAME).clear();
+
+        embeddedBroker.shutdown();
+    }
+
+    /**
+     * Tests Kafka streamer.
+     *
+     * @throws TimeoutException If timed out.
+     * @throws InterruptedException If interrupted.
+     */
+    @Test
+    public void testKafkaStreamer() throws TimeoutException, InterruptedException {
+        embeddedBroker.createTopic(TOPIC_NAME, PARTITIONS, REPLICATION_FACTOR);
+
+        Map<String, String> keyValMap = produceStream(TOPIC_NAME);
+
+        consumerStream(TOPIC_NAME, keyValMap);
+    }
+
+    /**
+     * Sends messages to Kafka.
+     *
+     * @param topic Topic name.
+     * @return Map of key value messages.
+     */
+    private Map<String, String> produceStream(String topic) {
+        // Generate random subnets.
+        List<Integer> subnet = new ArrayList<>();
+
+        for (int i = 1; i <= CNT; i++)
+            subnet.add(i);
+
+        Collections.shuffle(subnet);
+
+        List<ProducerRecord<String, String>> messages = new ArrayList<>(CNT);
+
+        Map<String, String> keyValMap = new HashMap<>();
+
+        for (int evt = 0; evt < CNT; evt++) {
+            long runtime = System.currentTimeMillis();
+
+            String ip = KEY_PREFIX + subnet.get(evt);
+
+            String msg = runtime + VALUE_URL + ip;
+
+            messages.add(new ProducerRecord<>(topic, ip, msg));
+
+            keyValMap.put(ip, msg);
+        }
+
+        embeddedBroker.sendMessages(messages);
+
+        return keyValMap;
+    }
+
+    /**
+     * Consumes Kafka stream via Ignite.
+     *
+     * @param topic Topic name.
+     * @param keyValMap Expected key value map.
+     * @throws TimeoutException If timed out.
+     * @throws InterruptedException If interrupted.
+     */
+    private void consumerStream(String topic, Map<String, String> keyValMap)
+        throws TimeoutException, InterruptedException {
+        KafkaStreamer<String, String> kafkaStmr = null;
+
+        Ignite ignite = grid();
+
+        try (IgniteDataStreamer<String, String> stmr = ignite.dataStreamer(DEFAULT_CACHE_NAME)) {
+            stmr.allowOverwrite(true);
+            stmr.autoFlushFrequency(10);
+
+            // Configure Kafka streamer.
+            kafkaStmr = new KafkaStreamer<>();
+
+            // Get the cache.
+            IgniteCache<String, String> cache = ignite.cache(DEFAULT_CACHE_NAME);
+
+            // Set Ignite instance.
+            kafkaStmr.setIgnite(ignite);
+
+            // Set data streamer instance.
+            kafkaStmr.setStreamer(stmr);
+
+            // Set the topic.
+            kafkaStmr.setTopic(Arrays.asList(topic));
+
+            // Set the number of threads.
+            kafkaStmr.setThreads(4);
+
+            // Set the consumer configuration.
+            kafkaStmr.setConsumerConfig(
+                createDefaultConsumerConfig(embeddedBroker.getBrokerAddress(), "groupX"));
+
+            kafkaStmr.setMultipleTupleExtractor(
+                record -> {
+                    Map<String, String> entries = new HashMap<>();
+
+                    try {
+                        String key = (String)record.key();
+                        String val = (String)record.value();
+
+                        // Convert the message into number of cache entries with same key or dynamic key from actual message.
+                        // For now using key as cache entry key and value as cache entry value - for test purpose.
+                        entries.put(key, val);
+                    }
+                    catch (Exception ex) {
+                        fail("Unexpected error." + ex);
+                    }
+
+                    return entries;
+                });
+
+            // Start kafka streamer.
+            kafkaStmr.start();
+
+            final CountDownLatch latch = new CountDownLatch(CNT);
+
+            IgniteBiPredicate<UUID, CacheEvent> locLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
+                @IgniteInstanceResource
+                private Ignite ig;
+
+                @LoggerResource
+                private IgniteLogger log;
+
+                /** {@inheritDoc} */
+                @Override public boolean apply(UUID uuid, CacheEvent evt) {
+                    latch.countDown();
+
+                    if (log.isInfoEnabled()) {
+                        IgniteEx igEx = (IgniteEx)ig;
+
+                        UUID nodeId = igEx.localNode().id();
+
+                        log.info("Recive event=" + evt + ", nodeId=" + nodeId);
+                    }
+
+                    return true;
+                }
+            };
+
+            ignite.events(ignite.cluster().forCacheNodes(DEFAULT_CACHE_NAME)).remoteListen(locLsnr, null, EVT_CACHE_OBJECT_PUT);
+
+            // Checks all events successfully processed in 10 seconds.
+            assertTrue("Failed to wait latch completion, still wait " + latch.getCount() + " events",
+                latch.await(10, TimeUnit.SECONDS));
+
+            for (Map.Entry<String, String> entry : keyValMap.entrySet())
+                assertEquals(entry.getValue(), cache.get(entry.getKey()));
+        }
+        finally {
+            if (kafkaStmr != null)
+                kafkaStmr.stop();
+        }
+    }
+
+    /**
+     * Creates default consumer config.
+     *
+     * @param servers Bootstrap servers' address in the form of &lt;server:port;server:port&gt;.
+     * @param grpId Group Id for kafka subscriber.
+     * @return Kafka consumer configuration.
+     */
+    private Properties createDefaultConsumerConfig(String servers, String grpId) {
+        A.notNull(servers, "bootstrap servers");
+        A.notNull(grpId, "groupId");
+
+        Properties props = new Properties();
+
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, grpId);
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+
+        return props;
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/TestKafkaBroker.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/TestKafkaBroker.java
new file mode 100644
index 0000000..24dbfc8
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/TestKafkaBroker.java
@@ -0,0 +1,237 @@
+/*
+ * 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 java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeoutException;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServer;
+import kafka.utils.TestUtils;
+import kafka.utils.ZkUtils;
+import kafka.zk.KafkaZkClient;
+import org.I0Itec.zkclient.ZkClient;
+import org.I0Itec.zkclient.ZkConnection;
+import org.apache.curator.test.TestingServer;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.SystemTime;
+import scala.Tuple2;
+
+/**
+ * Kafka Test Broker.
+ */
+public class TestKafkaBroker {
+    /** 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 final int ZK_PORT = 21811;
+
+    /** Broker host. */
+    private static final String BROKER_HOST = "localhost";
+
+    /** Broker port. */
+    private static final int BROKER_PORT = 11092;
+
+    /** Kafka config. */
+    private KafkaConfig kafkaCfg;
+
+    /** Kafka server. */
+    private KafkaServer kafkaSrv;
+
+    /** ZooKeeper. */
+    private TestingServer zkServer;
+
+    /** Kafka Zookeeper utils. */
+    private ZkUtils zkUtils;
+
+    /**
+     * Kafka broker constructor.
+     */
+    public TestKafkaBroker() {
+        try {
+            setupZooKeeper();
+
+            setupKafkaServer();
+        }
+        catch (Exception e) {
+            throw new RuntimeException("Failed to start Kafka: " + e);
+        }
+    }
+
+    /**
+     * Creates a topic.
+     *
+     * @param topic Topic name.
+     * @param partitions Number of partitions for the topic.
+     * @param replicationFactor Replication factor.
+     * @throws TimeoutException If operation is timed out.
+     * @throws InterruptedException If interrupted.
+     */
+    public void createTopic(String topic, int partitions, int replicationFactor)
+        throws TimeoutException, InterruptedException {
+        List<KafkaServer> servers = new ArrayList<>();
+
+        servers.add(kafkaSrv);
+
+        KafkaZkClient client = kafkaSrv.zkClient();
+
+        TestUtils.createTopic(client, topic, partitions, replicationFactor,
+            scala.collection.JavaConversions.asScalaBuffer(servers), new Properties());
+    }
+
+    /**
+     * Sends a message to Kafka broker.
+     *
+     * @param records List of records.
+     * @return Producer used to send the message.
+     */
+    public void sendMessages(List<ProducerRecord<String, String>> records) {
+        Producer<String, String> producer = new KafkaProducer<>(getProducerConfig());
+
+        for (ProducerRecord<String, String> rec : records)
+            producer.send(rec);
+
+        producer.flush();
+        producer.close();
+    }
+
+    /**
+     * Shuts down test Kafka broker.
+     */
+    public void shutdown() {
+        if (zkUtils != null)
+            zkUtils.close();
+
+        if (kafkaSrv != null)
+            kafkaSrv.shutdown();
+
+        if (zkServer != null) {
+            try {
+                zkServer.stop();
+            }
+            catch (IOException ignored) {
+                // No-op.
+            }
+        }
+
+        List<String> logDirs = scala.collection.JavaConversions.seqAsJavaList(kafkaCfg.logDirs());
+
+        for (String logDir : logDirs)
+            U.delete(new File(logDir));
+    }
+
+    /**
+     * Sets up test Kafka broker.
+     *
+     * @throws IOException If failed.
+     */
+    private void setupKafkaServer() throws IOException {
+        kafkaCfg = new KafkaConfig(getKafkaConfig());
+
+        kafkaSrv = TestUtils.createServer(kafkaCfg, new SystemTime());
+
+        kafkaSrv.startup();
+    }
+
+    /**
+     * Sets up ZooKeeper test server.
+     *
+     * @throws Exception If failed.
+     */
+    private void setupZooKeeper() throws Exception {
+        zkServer = new TestingServer(ZK_PORT, true);
+
+        Tuple2<ZkClient, ZkConnection> zkTuple = ZkUtils.createZkClientAndConnection(zkServer.getConnectString(),
+            ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT);
+
+        zkUtils = new ZkUtils(zkTuple._1(), zkTuple._2(), false);
+    }
+
+    /**
+     * Obtains Kafka config.
+     *
+     * @return Kafka config.
+     * @throws IOException If failed.
+     */
+    private Properties getKafkaConfig() throws IOException {
+        Properties props = new Properties();
+
+        props.put("broker.id", "0");
+        props.put("zookeeper.connect", zkServer.getConnectString());
+        props.put("host.name", BROKER_HOST);
+        props.put("port", BROKER_PORT);
+        props.put("offsets.topic.replication.factor", "1");
+        props.put("log.dir", createTmpDir("_cfg").getAbsolutePath());
+        props.put("log.flush.interval.messages", "1");
+        props.put("log.flush.interval.ms", "10");
+
+        return props;
+    }
+
+    /**
+     * Obtains broker address.
+     *
+     * @return Kafka broker address.
+     */
+    public String getBrokerAddress() {
+        return BROKER_HOST + ":" + BROKER_PORT;
+    }
+
+    /**
+     * Obtains producer config.
+     *
+     * @return Kafka Producer config.
+     */
+    private Properties getProducerConfig() {
+        Properties props = new Properties();
+
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getBrokerAddress());
+        props.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaTestProducer");
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+
+        return props;
+    }
+
+    /**
+     * Creates temporary directory.
+     *
+     * @param prefix Prefix.
+     * @return Created file.
+     * @throws IOException If failed.
+     */
+    private static File createTmpDir(String prefix) throws IOException {
+        Path path = Files.createTempDirectory(prefix);
+
+        return path.toFile();
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorMock.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorMock.java
new file mode 100644
index 0000000..4c912b9
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorMock.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.ignite.stream.kafka.connect;
+
+import org.apache.kafka.connect.connector.Task;
+
+/**
+ * Sink connector mock for tests for using the task mock.
+ */
+public class IgniteSinkConnectorMock extends IgniteSinkConnector {
+    /** {@inheritDoc} */
+    @Override public Class<? extends Task> taskClass() {
+        return IgniteSinkTaskMock.class;
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorTest.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorTest.java
new file mode 100644
index 0000000..8a70636
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorTest.java
@@ -0,0 +1,307 @@
+/*
+ * 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.connect;
+
+import java.lang.reflect.Field;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.stream.StreamSingleTupleExtractor;
+import org.apache.ignite.stream.kafka.TestKafkaBroker;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.utils.SystemTime;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
+import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.Worker;
+import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.runtime.isolation.Plugins;
+import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
+import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;
+import org.apache.kafka.connect.runtime.standalone.StandaloneHerder;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.apache.kafka.connect.storage.OffsetBackingStore;
+import org.apache.kafka.connect.util.Callback;
+import org.apache.kafka.connect.util.ConnectUtils;
+import org.apache.kafka.connect.util.FutureCallback;
+import org.junit.Test;
+
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
+import static org.easymock.EasyMock.mock;
+
+/**
+ * Tests for {@link IgniteSinkConnector}.
+ */
+public class IgniteSinkConnectorTest extends GridCommonAbstractTest {
+    /** Number of input messages. */
+    private static final int EVENT_CNT = 10000;
+
+    /** Cache name. */
+    private static final String CACHE_NAME = "testCache";
+
+    /** Test topics. */
+    private static final String[] TOPICS = {"sink-test1", "sink-test2"};
+
+    /** Kafka partition. */
+    private static final int PARTITIONS = 3;
+
+    /** Kafka replication factor. */
+    private static final int REPLICATION_FACTOR = 1;
+
+    /** Worker id. */
+    private static final String WORKER_ID = "workerId";
+
+    /** Test Kafka broker. */
+    private TestKafkaBroker kafkaBroker;
+
+    /** Worker to run tasks. */
+    private Worker worker;
+
+    /** Workers' herder. */
+    private Herder herder;
+
+    /** Ignite server node. */
+    private static Ignite grid;
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        kafkaBroker = new TestKafkaBroker();
+
+        for (String topic : TOPICS)
+            kafkaBroker.createTopic(topic, PARTITIONS, REPLICATION_FACTOR);
+
+        Map<String, String> props = makeWorkerProps();
+        WorkerConfig workerCfg = new StandaloneConfig(props);
+
+        OffsetBackingStore offBackingStore = mock(OffsetBackingStore.class);
+        offBackingStore.configure(workerCfg);
+
+        worker = new Worker(WORKER_ID, new SystemTime(), new Plugins(props), workerCfg, offBackingStore);
+        worker.start();
+
+        herder = new StandaloneHerder(worker, ConnectUtils.lookupKafkaClusterId(workerCfg));
+        herder.start();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        herder.stop();
+
+        worker.stop();
+
+        kafkaBroker.shutdown();
+
+        grid.cache(CACHE_NAME).removeAll();
+
+        // reset cache name to overwrite task configurations.
+        Field field = IgniteSinkTask.class.getDeclaredField("cacheName");
+
+        field.setAccessible(true);
+        field.set(IgniteSinkTask.class, null);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        IgniteConfiguration cfg = loadConfiguration("example-ignite.xml");
+
+        cfg.setClientMode(false);
+
+        grid = startGrid("igniteServerNode", cfg);
+    }
+
+    /**
+     * @throws Exception if failed.
+     */
+    @Test
+    public void testSinkPutsWithoutTransformation() throws Exception {
+        Map<String, String> sinkProps = makeSinkProps(Utils.join(TOPICS, ","));
+
+        sinkProps.remove(IgniteSinkConstants.SINGLE_TUPLE_EXTRACTOR_CLASS);
+
+        testSinkPuts(sinkProps, false);
+    }
+
+    /**
+     * @throws Exception if failed.
+     */
+    @Test
+    public void testSinkPutsWithTransformation() throws Exception {
+        testSinkPuts(makeSinkProps(Utils.join(TOPICS, ",")), true);
+    }
+
+    /**
+     * Tests the whole data flow from injecting data to Kafka to transferring it to the grid. It reads from two
+     * specified Kafka topics, because a sink task can read from multiple topics.
+     *
+     * @param sinkProps Sink properties.
+     * @param keyless Tests on Kafka stream with null keys if true.
+     * @throws Exception Thrown in case of the failure.
+     */
+    private void testSinkPuts(Map<String, String> sinkProps, boolean keyless) throws Exception {
+        FutureCallback<Herder.Created<ConnectorInfo>> cb = new FutureCallback<>(new Callback<Herder.Created<ConnectorInfo>>() {
+            @Override public void onCompletion(Throwable error, Herder.Created<ConnectorInfo> info) {
+                if (error != null)
+                    throw new RuntimeException("Failed to create a job!");
+            }
+        });
+
+        herder.putConnectorConfig(
+            sinkProps.get(ConnectorConfig.NAME_CONFIG),
+            sinkProps, false, cb);
+
+        cb.get();
+
+        final CountDownLatch latch = new CountDownLatch(EVENT_CNT * TOPICS.length);
+
+        final IgnitePredicate<Event> putLsnr = new IgnitePredicate<Event>() {
+            @Override public boolean apply(Event evt) {
+                assert evt != null;
+
+                latch.countDown();
+
+                return true;
+            }
+        };
+
+        grid.events(grid.cluster().forCacheNodes(CACHE_NAME)).localListen(putLsnr, EVT_CACHE_OBJECT_PUT);
+
+        IgniteCache<String, String> cache = grid.cache(CACHE_NAME);
+
+        assertEquals(0, cache.size(CachePeekMode.PRIMARY));
+
+        Map<String, String> keyValMap = new HashMap<>(EVENT_CNT * TOPICS.length);
+
+        // Produces events for the specified number of topics
+        for (String topic : TOPICS)
+            keyValMap.putAll(produceStream(topic, keyless));
+
+        // Checks all events successfully processed in 10 seconds.
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+        grid.events(grid.cluster().forCacheNodes(CACHE_NAME)).stopLocalListen(putLsnr);
+
+        // Checks that each event was processed properly.
+        for (Map.Entry<String, String> entry : keyValMap.entrySet())
+            assertEquals(entry.getValue(), cache.get(entry.getKey()));
+
+        assertEquals(EVENT_CNT * TOPICS.length, cache.size(CachePeekMode.PRIMARY));
+    }
+
+    /**
+     * Sends messages to Kafka.
+     *
+     * @param topic Topic name.
+     * @param keyless Indicates whether a Kafka key is specified.
+     * @return Map of key value messages.
+     */
+    private Map<String, String> produceStream(String topic, boolean keyless) {
+        List<ProducerRecord<String, String>> messages = new ArrayList<>(EVENT_CNT);
+
+        Map<String, String> keyValMap = new HashMap<>();
+
+        for (int evt = 0; evt < EVENT_CNT; evt++) {
+            long runtime = System.currentTimeMillis();
+
+            String key = null;
+            if (!keyless)
+                key = topic + ":" + String.valueOf(evt);
+
+            String msg = topic + ":" + String.valueOf(evt) + "_" + runtime;
+
+            messages.add(new ProducerRecord<>(topic, key, msg));
+
+            if (!keyless)
+                keyValMap.put(key, msg);
+            else
+                keyValMap.put(topic + ":" + String.valueOf(evt), String.valueOf(runtime));
+        }
+
+        kafkaBroker.sendMessages(messages);
+
+        return keyValMap;
+    }
+
+    /**
+     * Creates properties for test sink connector.
+     *
+     * @param topics Topics.
+     * @return Test sink connector properties.
+     */
+    private Map<String, String> makeSinkProps(String topics) {
+        Map<String, String> props = new HashMap<>();
+
+        props.put(SinkConnector.TOPICS_CONFIG, topics);
+        props.put(ConnectorConfig.TASKS_MAX_CONFIG, "2");
+        props.put(ConnectorConfig.NAME_CONFIG, "test-sink-connector");
+        props.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, IgniteSinkConnectorMock.class.getName());
+        props.put(IgniteSinkConstants.CACHE_NAME, "testCache");
+        props.put(IgniteSinkConstants.CACHE_ALLOW_OVERWRITE, "true");
+        props.put(IgniteSinkConstants.CACHE_CFG_PATH, "example-ignite.xml");
+        props.put(IgniteSinkConstants.SINGLE_TUPLE_EXTRACTOR_CLASS,
+            "org.apache.ignite.stream.kafka.connect.IgniteSinkConnectorTest$TestExtractor");
+
+        return props;
+    }
+
+    /**
+     * Creates properties for Kafka Connect workers.
+     *
+     * @return Worker configurations.
+     */
+    private Map<String, String> makeWorkerProps() {
+        Map<String, String> props = new HashMap<>();
+
+        props.put(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
+        props.put(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
+        props.put("internal.key.converter.schemas.enable", "false");
+        props.put("internal.value.converter.schemas.enable", "false");
+        props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
+        props.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
+        props.put("key.converter.schemas.enable", "false");
+        props.put("value.converter.schemas.enable", "false");
+        props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker.getBrokerAddress());
+        props.put("offset.storage.file.filename", "/tmp/connect.offsets");
+        // fast flushing for testing.
+        props.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "10");
+
+        return props;
+    }
+
+    /**
+     * Test transformer.
+     */
+    static class TestExtractor implements StreamSingleTupleExtractor<SinkRecord, String, String> {
+        /** {@inheritDoc} */
+        @Override public Map.Entry<String, String> extract(SinkRecord msg) {
+            String[] parts = ((String)msg.value()).split("_");
+            return new AbstractMap.SimpleEntry<String, String>(parts[0], parts[1]);
+        }
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkTaskMock.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkTaskMock.java
new file mode 100644
index 0000000..58a59b5
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkTaskMock.java
@@ -0,0 +1,29 @@
+/*
+ * 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.connect;
+
+/**
+ * Sink task mock for tests. It avoids closing the grid from test to test.
+ */
+public class IgniteSinkTaskMock extends IgniteSinkTask {
+    /** {@inheritDoc} */
+    @Override public void stop() {
+        // Don't stop the grid for tests.
+        setStopped(true);
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorMock.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorMock.java
new file mode 100644
index 0000000..0157a17
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorMock.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.ignite.stream.kafka.connect;
+
+import org.apache.kafka.connect.connector.Task;
+
+/**
+ * Source connector mock for tests for using the task mock.
+ */
+public class IgniteSourceConnectorMock extends IgniteSourceConnector {
+    /** {@inheritDoc} */
+    @Override public Class<? extends Task> taskClass() {
+        return IgniteSourceTaskMock.class;
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
new file mode 100644
index 0000000..a1109e7
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
@@ -0,0 +1,355 @@
+/*
+ * 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.connect;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.stream.kafka.TestKafkaBroker;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.utils.SystemTime;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
+import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.Worker;
+import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.runtime.isolation.Plugins;
+import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
+import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;
+import org.apache.kafka.connect.runtime.standalone.StandaloneHerder;
+import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
+import org.apache.kafka.connect.util.Callback;
+import org.apache.kafka.connect.util.ConnectUtils;
+import org.apache.kafka.connect.util.FutureCallback;
+import org.junit.Test;
+
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
+
+/**
+ * Tests for {@link IgniteSourceConnector}.
+ */
+public class IgniteSourceConnectorTest extends GridCommonAbstractTest {
+    /** Number of input messages. */
+    private static final int EVENT_CNT = 100;
+
+    /** Cache name. */
+    private static final String CACHE_NAME = "testCache";
+
+    /** Test topics created by connector. */
+    private static final String[] TOPICS = {"src-test1", "src-test2"};
+
+    /** Worker id. */
+    private static final String WORKER_ID = "workerId";
+
+    /** Test Kafka broker. */
+    private TestKafkaBroker kafkaBroker;
+
+    /** Worker to run tasks. */
+    private Worker worker;
+
+    /** Workers' herder. */
+    private Herder herder;
+
+    /** Ignite server node shared among tests. */
+    private static Ignite grid;
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        IgniteConfiguration cfg = loadConfiguration("example-ignite.xml");
+
+        cfg.setClientMode(false);
+
+        grid = startGrid("igniteServerNode", cfg);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        kafkaBroker = new TestKafkaBroker();
+
+        Map<String, String> props = makeWorkerProps();
+        WorkerConfig workerCfg = new StandaloneConfig(props);
+
+        MemoryOffsetBackingStore offBackingStore = new MemoryOffsetBackingStore();
+        offBackingStore.configure(workerCfg);
+
+        worker = new Worker(WORKER_ID, new SystemTime(), new Plugins(props), workerCfg, offBackingStore);
+        worker.start();
+
+        herder = new StandaloneHerder(worker, ConnectUtils.lookupKafkaClusterId(workerCfg));
+        herder.start();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        herder.stop();
+
+        worker.stop();
+
+        kafkaBroker.shutdown();
+
+        grid.cache(CACHE_NAME).clear();
+
+        // reset cache name to overwrite task configurations.
+        Field field = IgniteSourceTask.class.getDeclaredField("cacheName");
+
+        field.setAccessible(true);
+        field.set(IgniteSourceTask.class, null);
+    }
+
+    /**
+     * Tests data flow from injecting data into grid and transferring it to Kafka cluster
+     * without user-specified filter.
+     *
+     * @throws Exception Thrown in case of the failure.
+     */
+    @Test
+    public void testEventsInjectedIntoKafkaWithoutFilter() throws Exception {
+        Map<String, String> srcProps = makeSourceProps(Utils.join(TOPICS, ","));
+
+        srcProps.remove(IgniteSourceConstants.CACHE_FILTER_CLASS);
+
+        doTest(srcProps, false);
+    }
+
+    /**
+     * Tests data flow from injecting data into grid and transferring it to Kafka cluster.
+     *
+     * @throws Exception Thrown in case of the failure.
+     */
+    @Test
+    public void testEventsInjectedIntoKafka() throws Exception {
+        doTest(makeSourceProps(Utils.join(TOPICS, ",")), true);
+    }
+
+    /**
+     * Tests the source with the specified source configurations.
+     *
+     * @param srcProps Source properties.
+     * @param conditioned Flag indicating whether filtering is enabled.
+     * @throws Exception Fails if error.
+     */
+    private void doTest(Map<String, String> srcProps, boolean conditioned) throws Exception {
+        FutureCallback<Herder.Created<ConnectorInfo>> cb = new FutureCallback<>(new Callback<Herder.Created<ConnectorInfo>>() {
+            @Override public void onCompletion(Throwable error, Herder.Created<ConnectorInfo> info) {
+                if (error != null)
+                    throw new RuntimeException("Failed to create a job!", error);
+            }
+        });
+
+        herder.putConnectorConfig(
+            srcProps.get(ConnectorConfig.NAME_CONFIG),
+            srcProps, true, cb);
+
+        cb.get();
+
+        // Ugh! To be sure Kafka Connect's worker thread is properly started...
+        Thread.sleep(5000);
+
+        final CountDownLatch latch = new CountDownLatch(EVENT_CNT);
+
+        final IgnitePredicate<CacheEvent> locLsnr = new IgnitePredicate<CacheEvent>() {
+            @Override public boolean apply(CacheEvent evt) {
+                assert evt != null;
+
+                latch.countDown();
+
+                return true;
+            }
+        };
+
+        grid.events(grid.cluster().forCacheNodes(CACHE_NAME)).localListen(locLsnr, EVT_CACHE_OBJECT_PUT);
+
+        IgniteCache<String, String> cache = grid.cache(CACHE_NAME);
+
+        assertEquals(0, cache.size(CachePeekMode.PRIMARY));
+
+        Map<String, String> keyValMap = new HashMap<>(EVENT_CNT);
+
+        keyValMap.putAll(sendData());
+
+        // Checks all events are processed.
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+        grid.events(grid.cluster().forCacheNodes(CACHE_NAME)).stopLocalListen(locLsnr);
+
+        assertEquals(EVENT_CNT, cache.size(CachePeekMode.PRIMARY));
+
+        // Checks the events are transferred to Kafka broker.
+        if (conditioned)
+            checkDataDelivered(EVENT_CNT * TOPICS.length / 2);
+        else
+            checkDataDelivered(EVENT_CNT * TOPICS.length);
+
+    }
+
+    /**
+     * Sends messages to the grid.
+     *
+     * @return Map of key value messages.
+     * @throws IOException If failed.
+     */
+    private Map<String, String> sendData() throws IOException {
+        Map<String, String> keyValMap = new HashMap<>();
+
+        for (int evt = 0; evt < EVENT_CNT; evt++) {
+            long runtime = System.currentTimeMillis();
+
+            String key = "test_" + String.valueOf(evt);
+            String msg = runtime + String.valueOf(evt);
+
+            if (evt >= EVENT_CNT / 2)
+                key = "conditioned_" + key;
+
+            grid.cache(CACHE_NAME).put(key, msg);
+
+            keyValMap.put(key, msg);
+        }
+
+        return keyValMap;
+    }
+
+    /**
+     * Checks if events were delivered to Kafka server.
+     *
+     * @param expectedEventsCnt Expected events count.
+     * @throws Exception If failed.
+     */
+    private void checkDataDelivered(final int expectedEventsCnt) throws Exception {
+        Properties props = new Properties();
+
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker.getBrokerAddress());
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-grp");
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
+        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 20000);
+        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000);
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+            "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+            "org.apache.ignite.stream.kafka.connect.serialization.CacheEventDeserializer");
+
+        final KafkaConsumer<String, CacheEvent> consumer = new KafkaConsumer<>(props);
+
+        consumer.subscribe(Arrays.asList(TOPICS));
+
+        final AtomicInteger evtCnt = new AtomicInteger();
+
+        try {
+            // Wait for expected events count.
+            GridTestUtils.waitForCondition(new GridAbsPredicate() {
+                @Override public boolean apply() {
+                    ConsumerRecords<String, CacheEvent> records = consumer.poll(10);
+                    for (ConsumerRecord<String, CacheEvent> record : records) {
+                        info("Record: " + record);
+
+                        evtCnt.getAndIncrement();
+                    }
+                    return evtCnt.get() >= expectedEventsCnt;
+                }
+            }, 20_000);
+
+            info("Waiting for unexpected records for 5 secs.");
+
+            assertFalse(GridTestUtils.waitForCondition(new GridAbsPredicate() {
+                @Override public boolean apply() {
+                    ConsumerRecords<String, CacheEvent> records = consumer.poll(10);
+                    for (ConsumerRecord<String, CacheEvent> record : records) {
+                        error("Unexpected record: " + record);
+
+                        evtCnt.getAndIncrement();
+                    }
+                    return evtCnt.get() > expectedEventsCnt;
+                }
+            }, 5_000));
+        }
+        catch (WakeupException ignored) {
+            // ignore for shutdown.
+        }
+        finally {
+            consumer.close();
+
+            assertEquals(expectedEventsCnt, evtCnt.get());
+        }
+    }
+
+    /**
+     * Creates properties for test source connector.
+     *
+     * @param topics Topics.
+     * @return Test source connector properties.
+     */
+    private Map<String, String> makeSourceProps(String topics) {
+        Map<String, String> props = new HashMap<>();
+
+        props.put(ConnectorConfig.TASKS_MAX_CONFIG, "2");
+        props.put(ConnectorConfig.NAME_CONFIG, "test-src-connector");
+        props.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, IgniteSourceConnectorMock.class.getName());
+        props.put(IgniteSourceConstants.CACHE_NAME, "testCache");
+        props.put(IgniteSourceConstants.CACHE_CFG_PATH, "example-ignite.xml");
+        props.put(IgniteSourceConstants.TOPIC_NAMES, topics);
+        props.put(IgniteSourceConstants.CACHE_EVENTS, "put");
+        props.put(IgniteSourceConstants.CACHE_FILTER_CLASS, TestCacheEventFilter.class.getName());
+        props.put(IgniteSourceConstants.INTL_BUF_SIZE, "1000000");
+
+        return props;
+    }
+
+    /**
+     * Creates properties for Kafka Connect workers.
+     *
+     * @return Worker configurations.
+     * @throws IOException If failed.
+     */
+    private Map<String, String> makeWorkerProps() throws IOException {
+        Map<String, String> props = new HashMap<>();
+
+        props.put(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
+        props.put(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
+        props.put("internal.key.converter.schemas.enable", "false");
+        props.put("internal.value.converter.schemas.enable", "false");
+        props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
+        props.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.ignite.stream.kafka.connect.serialization.CacheEventConverter");
+        props.put("key.converter.schemas.enable", "false");
+        props.put("value.converter.schemas.enable", "false");
+        props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker.getBrokerAddress());
+        props.put("offset.storage.file.filename", "/tmp/connect.offsets");
+        // fast flushing for testing.
+        props.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "10");
+
+        return props;
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceTaskMock.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceTaskMock.java
new file mode 100644
index 0000000..4535ad5
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceTaskMock.java
@@ -0,0 +1,31 @@
+/*
+ * 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.connect;
+
+/**
+ * Source task mock for tests. It avoids closing the grid from test to test.
+ */
+public class IgniteSourceTaskMock extends IgniteSourceTask {
+    /** {@inheritDoc} */
+    @Override public void stop() {
+        // Don't stop the grid for tests.
+        stopRemoteListen();
+
+        setStopped(true);
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/TestCacheEventFilter.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/TestCacheEventFilter.java
new file mode 100644
index 0000000..bf28a2b
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/TestCacheEventFilter.java
@@ -0,0 +1,31 @@
+/*
+ * 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.connect;
+
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.lang.IgnitePredicate;
+
+/**
+ * Test user-defined filter.
+ */
+class TestCacheEventFilter implements IgnitePredicate<CacheEvent> {
+    /** {@inheritDoc} */
+    @Override public boolean apply(CacheEvent evt) {
+        return ((String)evt.key()).startsWith("conditioned_");
+    }
+}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/package-info.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/package-info.java
new file mode 100644
index 0000000..d668d30
--- /dev/null
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/package-info.java
@@ -0,0 +1,22 @@
+/*
+ *  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.
+ */
+
+/**
+ * Contains implementation of Kafka Streamer.
+ */
+
+package org.apache.ignite.stream.kafka;
diff --git a/modules/kafka-ext/src/test/resources/example-ignite.xml b/modules/kafka-ext/src/test/resources/example-ignite.xml
new file mode 100644
index 0000000..f23a306
--- /dev/null
+++ b/modules/kafka-ext/src/test/resources/example-ignite.xml
@@ -0,0 +1,73 @@
+<?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.
+-->
+
+<!--
+    Ignite configuration with all defaults and enabled p2p deployment and enabled events.
+    Used for testing IgniteSink running Ignite in a client mode.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:util="http://www.springframework.org/schema/util"
+       xsi:schemaLocation="
+        http://www.springframework.org/schema/beans
+        http://www.springframework.org/schema/beans/spring-beans.xsd
+        http://www.springframework.org/schema/util
+        http://www.springframework.org/schema/util/spring-util.xsd">
+    <bean id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
+        <!-- Enable peer class loading for remote events. -->
+        <property name="peerClassLoadingEnabled" value="true"/>
+        <!-- Enable client mode. -->
+        <property name="clientMode" value="true"/>
+
+        <!-- Cache accessed from IgniteSink. -->
+        <property name="cacheConfiguration">
+            <list>
+                <!-- Partitioned cache example configuration with configurations adjusted to server nodes'. -->
+                <bean class="org.apache.ignite.configuration.CacheConfiguration">
+                    <property name="atomicityMode" value="ATOMIC"/>
+
+                    <property name="name" value="testCache"/>
+                </bean>
+            </list>
+        </property>
+
+        <!-- Enable cache events. -->
+        <property name="includeEventTypes">
+            <list>
+                <!-- Cache events. -->
+                <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
+            </list>
+        </property>
+
+        <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
+        <property name="discoverySpi">
+            <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
+                <property name="ipFinder">
+                    <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
+                        <property name="addresses">
+                            <list>
+                                <value>127.0.0.1:47500</value>
+                            </list>
+                        </property>
+                    </bean>
+                </property>
+            </bean>
+        </property>
+    </bean>
+</beans>
diff --git a/pom.xml b/pom.xml
index 2aa8af9..55d3af5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,6 +54,7 @@
         <module>modules/storm-ext</module>
         <module>modules/camel-ext</module>
         <module>modules/jms11-ext</module>
+        <module>modules/kafka-ext</module>
     </modules>
 
     <profiles>