diff --git a/build.gradle b/build.gradle
new file mode 100644
index 0000000..d694dfa
--- /dev/null
+++ b/build.gradle
@@ -0,0 +1,24 @@
+plugins {
+    id 'java'
+}
+
+group 'org.apache.geode'
+version '1.0-SNAPSHOT'
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+
+    compile 'org.apache.geode:geode-core:1.11.0'
+    compile(group: 'org.apache.kafka', name: 'connect-api', version: '2.3.1')
+    testCompile(group: 'org.apache.kafka', name: 'kafka_2.12', version: '2.3.1')
+    testCompile(group: 'org.apache.kafka', name: 'kafka-streams-test-utils', version: '1.1.0')
+    testCompile(group: 'org.apache.curator', name: 'curator-framework', version: '4.2.0')
+    testCompile(group: 'org.apache.kafka', name: 'connect-runtime', version: '2.3.1')
+
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+}
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
new file mode 100644
index 0000000..87b738c
--- /dev/null
+++ b/gradle/wrapper/gradle-wrapper.jar
Binary files differ
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
new file mode 100644
index 0000000..44e7c4d
--- /dev/null
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -0,0 +1,5 @@
+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-5.2.1-bin.zip
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
new file mode 100755
index 0000000..af6708f
--- /dev/null
+++ b/gradlew
@@ -0,0 +1,172 @@
+#!/usr/bin/env sh
+
+##############################################################################
+##
+##  Gradle start up script for UN*X
+##
+##############################################################################
+
+# Attempt to set APP_HOME
+# Resolve links: $0 may be a link
+PRG="$0"
+# Need this for relative symlinks.
+while [ -h "$PRG" ] ; do
+    ls=`ls -ld "$PRG"`
+    link=`expr "$ls" : '.*-> \(.*\)$'`
+    if expr "$link" : '/.*' > /dev/null; then
+        PRG="$link"
+    else
+        PRG=`dirname "$PRG"`"/$link"
+    fi
+done
+SAVED="`pwd`"
+cd "`dirname \"$PRG\"`/" >/dev/null
+APP_HOME="`pwd -P`"
+cd "$SAVED" >/dev/null
+
+APP_NAME="Gradle"
+APP_BASE_NAME=`basename "$0"`
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m"'
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD="maximum"
+
+warn () {
+    echo "$*"
+}
+
+die () {
+    echo
+    echo "$*"
+    echo
+    exit 1
+}
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+nonstop=false
+case "`uname`" in
+  CYGWIN* )
+    cygwin=true
+    ;;
+  Darwin* )
+    darwin=true
+    ;;
+  MINGW* )
+    msys=true
+    ;;
+  NONSTOP* )
+    nonstop=true
+    ;;
+esac
+
+CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+        # IBM's JDK on AIX uses strange locations for the executables
+        JAVACMD="$JAVA_HOME/jre/sh/java"
+    else
+        JAVACMD="$JAVA_HOME/bin/java"
+    fi
+    if [ ! -x "$JAVACMD" ] ; then
+        die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+    fi
+else
+    JAVACMD="java"
+    which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+fi
+
+# Increase the maximum file descriptors if we can.
+if [ "$cygwin" = "false" -a "$darwin" = "false" -a "$nonstop" = "false" ] ; then
+    MAX_FD_LIMIT=`ulimit -H -n`
+    if [ $? -eq 0 ] ; then
+        if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then
+            MAX_FD="$MAX_FD_LIMIT"
+        fi
+        ulimit -n $MAX_FD
+        if [ $? -ne 0 ] ; then
+            warn "Could not set maximum file descriptor limit: $MAX_FD"
+        fi
+    else
+        warn "Could not query maximum file descriptor limit: $MAX_FD_LIMIT"
+    fi
+fi
+
+# For Darwin, add options to specify how the application appears in the dock
+if $darwin; then
+    GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\""
+fi
+
+# For Cygwin, switch paths to Windows format before running java
+if $cygwin ; then
+    APP_HOME=`cygpath --path --mixed "$APP_HOME"`
+    CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+    JAVACMD=`cygpath --unix "$JAVACMD"`
+
+    # We build the pattern for arguments to be converted via cygpath
+    ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null`
+    SEP=""
+    for dir in $ROOTDIRSRAW ; do
+        ROOTDIRS="$ROOTDIRS$SEP$dir"
+        SEP="|"
+    done
+    OURCYGPATTERN="(^($ROOTDIRS))"
+    # Add a user-defined pattern to the cygpath arguments
+    if [ "$GRADLE_CYGPATTERN" != "" ] ; then
+        OURCYGPATTERN="$OURCYGPATTERN|($GRADLE_CYGPATTERN)"
+    fi
+    # Now convert the arguments - kludge to limit ourselves to /bin/sh
+    i=0
+    for arg in "$@" ; do
+        CHECK=`echo "$arg"|egrep -c "$OURCYGPATTERN" -`
+        CHECK2=`echo "$arg"|egrep -c "^-"`                                 ### Determine if an option
+
+        if [ $CHECK -ne 0 ] && [ $CHECK2 -eq 0 ] ; then                    ### Added a condition
+            eval `echo args$i`=`cygpath --path --ignore --mixed "$arg"`
+        else
+            eval `echo args$i`="\"$arg\""
+        fi
+        i=$((i+1))
+    done
+    case $i in
+        (0) set -- ;;
+        (1) set -- "$args0" ;;
+        (2) set -- "$args0" "$args1" ;;
+        (3) set -- "$args0" "$args1" "$args2" ;;
+        (4) set -- "$args0" "$args1" "$args2" "$args3" ;;
+        (5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;;
+        (6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;;
+        (7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;;
+        (8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;;
+        (9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;;
+    esac
+fi
+
+# Escape application args
+save () {
+    for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
+    echo " "
+}
+APP_ARGS=$(save "$@")
+
+# Collect all arguments for the java command, following the shell quoting and substitution rules
+eval set -- $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS "\"-Dorg.gradle.appname=$APP_BASE_NAME\"" -classpath "\"$CLASSPATH\"" org.gradle.wrapper.GradleWrapperMain "$APP_ARGS"
+
+# by default we should be in the correct project dir, but when run from Finder on Mac, the cwd is wrong
+if [ "$(uname)" = "Darwin" ] && [ "$HOME" = "$PWD" ]; then
+  cd "$(dirname "$0")"
+fi
+
+exec "$JAVACMD" "$@"
diff --git a/gradlew.bat b/gradlew.bat
new file mode 100644
index 0000000..0f8d593
--- /dev/null
+++ b/gradlew.bat
@@ -0,0 +1,84 @@
+@if "%DEBUG%" == "" @echo off
+@rem ##########################################################################
+@rem
+@rem  Gradle startup script for Windows
+@rem
+@rem ##########################################################################
+
+@rem Set local scope for the variables with windows NT shell
+if "%OS%"=="Windows_NT" setlocal
+
+set DIRNAME=%~dp0
+if "%DIRNAME%" == "" set DIRNAME=.
+set APP_BASE_NAME=%~n0
+set APP_HOME=%DIRNAME%
+
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+set DEFAULT_JVM_OPTS="-Xmx64m"
+
+@rem Find java.exe
+if defined JAVA_HOME goto findJavaFromJavaHome
+
+set JAVA_EXE=java.exe
+%JAVA_EXE% -version >NUL 2>&1
+if "%ERRORLEVEL%" == "0" goto init
+
+echo.
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:findJavaFromJavaHome
+set JAVA_HOME=%JAVA_HOME:"=%
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe
+
+if exist "%JAVA_EXE%" goto init
+
+echo.
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:init
+@rem Get command-line arguments, handling Windows variants
+
+if not "%OS%" == "Windows_NT" goto win9xME_args
+
+:win9xME_args
+@rem Slurp the command line arguments.
+set CMD_LINE_ARGS=
+set _SKIP=2
+
+:win9xME_args_slurp
+if "x%~1" == "x" goto execute
+
+set CMD_LINE_ARGS=%*
+
+:execute
+@rem Setup the command line
+
+set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
+
+@rem Execute Gradle
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS%
+
+:end
+@rem End local scope for the variables with windows NT shell
+if "%ERRORLEVEL%"=="0" goto mainEnd
+
+:fail
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
+rem the _cmd.exe /c_ return code!
+if  not "" == "%GRADLE_EXIT_CONSOLE%" exit 1
+exit /b 1
+
+:mainEnd
+if "%OS%"=="Windows_NT" endlocal
+
+:omega
diff --git a/settings.gradle b/settings.gradle
new file mode 100644
index 0000000..b5a2326
--- /dev/null
+++ b/settings.gradle
@@ -0,0 +1,2 @@
+rootProject.name = 'geode-kafka-connector'
+
diff --git a/src/main/java/kafka/GeodeKafkaSink.java b/src/main/java/kafka/GeodeKafkaSink.java
new file mode 100644
index 0000000..af3a22a
--- /dev/null
+++ b/src/main/java/kafka/GeodeKafkaSink.java
@@ -0,0 +1,5 @@
+package kafka;
+
+public class GeodeKafkaSink {
+
+}
diff --git a/src/main/java/kafka/GeodeKafkaSource.java b/src/main/java/kafka/GeodeKafkaSource.java
new file mode 100644
index 0000000..d1599e9
--- /dev/null
+++ b/src/main/java/kafka/GeodeKafkaSource.java
@@ -0,0 +1,62 @@
+package kafka;
+
+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.source.SourceConnector;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class GeodeKafkaSource extends SourceConnector {
+
+  public static String REGION_NAME = "GEODE_REGION_NAME";
+  private String regionName;
+  private static String TOPICS = "TOPICS";
+
+  private Map<String, String> sharedProps;
+  private static final ConfigDef CONFIG_DEF = new ConfigDef();
+
+
+  @Override
+  public Class<? extends Task> taskClass() {
+    return GeodeKafkaSourceTask.class;
+  }
+
+  @Override
+  public List<Map<String, String>> taskConfigs(int maxTasks) {
+    List<Map<String, String>> taskConfigs = new ArrayList<>();
+    Map<String, String> taskProps = new HashMap<>();
+
+    taskProps.putAll(sharedProps);
+
+    // use the same props for all tasks at the moment
+    for (int i = 0; i < maxTasks; i++)
+      taskConfigs.add(taskProps);
+
+    return taskConfigs;
+  }
+
+
+  @Override
+  public ConfigDef config() {
+    return CONFIG_DEF;
+  }
+
+  @Override
+  public void start(Map<String, String> props) {
+    sharedProps = props;
+  }
+
+  @Override
+  public void stop() {
+
+  }
+
+  @Override
+  public String version() {
+    return AppInfoParser.getVersion();
+  }
+}
diff --git a/src/main/java/kafka/GeodeKafkaSourceTask.java b/src/main/java/kafka/GeodeKafkaSourceTask.java
new file mode 100644
index 0000000..6492199
--- /dev/null
+++ b/src/main/java/kafka/GeodeKafkaSourceTask.java
@@ -0,0 +1,120 @@
+package kafka;
+
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.cache.query.CqAttributes;
+import org.apache.geode.cache.query.CqAttributesFactory;
+import org.apache.geode.cache.query.CqEvent;
+import org.apache.geode.cache.query.CqException;
+import org.apache.geode.cache.query.CqExistsException;
+import org.apache.geode.cache.query.CqListener;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.source.SourceTask;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class GeodeKafkaSourceTask extends SourceTask {
+  private static String REGION_NAME = "REGION_NAME";
+  private static String OFFSET = "OFFSET";
+  private static String topics[];
+  private int batchSize;
+  private int queueSize;
+  private static BlockingQueue<CqEvent> eventBuffer;
+  private Map<String, String> sourcePartition;
+  private Map<String, Long> offset;
+
+  private ClientCache clientCache;
+
+  @Override
+  public String version() {
+    return null;
+  }
+
+  @Override
+  public void start(Map<String, String> props) {
+    batchSize = 100;
+    queueSize = 100000;
+    String regionName = "someRegion";
+    eventBuffer = new LinkedBlockingQueue<>(queueSize);
+    topics = new String[] {"default"};
+    sourcePartition = new HashMap<>();
+    sourcePartition.put(REGION_NAME, regionName);
+
+    offset = new HashMap<>();
+    offset.put("OFFSET", 0L);
+
+    installOnGeode("localHost", 18888, "someRegion");
+  }
+
+  @Override
+  public List<SourceRecord> poll() throws InterruptedException {
+    ArrayList<SourceRecord> records = new ArrayList<>(batchSize);
+    ArrayList<CqEvent> events = new ArrayList<>(batchSize);
+    if (eventBuffer.drainTo(events, batchSize) > 0) {
+      for (CqEvent event : events) {
+
+        for (String topic : topics)
+          records.add(new SourceRecord(sourcePartition, offset, topic, null, event));
+      }
+
+      return records;
+    }
+
+    return null;
+  }
+
+  @Override
+  public void stop() {
+    clientCache.close(true);
+  }
+
+  private void installOnGeode(String locatorHost, int locatorPort, String regionName) {
+    clientCache = new ClientCacheFactory().set("durable-client-id", "someClient")
+        .set("durable-client-timeout", "200")
+        .setPoolSubscriptionEnabled(true).addPoolLocator(locatorHost, locatorPort).create();
+    CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
+    cqAttributesFactory.addCqListener(new GeodeKafkaSourceListener());
+    CqAttributes cqAttributes = cqAttributesFactory.create();
+    try {
+      clientCache.getQueryService().newCq("kafkaCQFor" + regionName, "select * from /" + regionName, cqAttributes,
+          true);
+    } catch (CqExistsException e) {
+      e.printStackTrace();
+    } catch (CqException e) {
+      e.printStackTrace();
+    }
+    clientCache.readyForEvents();
+  }
+
+  private static class GeodeKafkaSourceListener implements CqListener {
+
+    @Override
+    public void onEvent(CqEvent aCqEvent) {
+      try {
+        eventBuffer.offer(aCqEvent, 2, TimeUnit.SECONDS);
+      } catch (InterruptedException e) {
+
+        while (true) {
+          try {
+            if (!eventBuffer.offer(aCqEvent, 2, TimeUnit.SECONDS))
+              break;
+          } catch (InterruptedException ex) {
+            ex.printStackTrace();
+          }
+          System.out.println("GeodeKafkaSource Queue is full");
+        }
+      }
+    }
+
+    @Override
+    public void onError(CqEvent aCqEvent) {
+
+    }
+  }
+}
diff --git a/src/test/java/kafka/GeodeKafkaTestCluster.java b/src/test/java/kafka/GeodeKafkaTestCluster.java
new file mode 100644
index 0000000..fbaba2c
--- /dev/null
+++ b/src/test/java/kafka/GeodeKafkaTestCluster.java
@@ -0,0 +1,128 @@
+package kafka;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.cache.client.ClientRegionShortcut;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public class GeodeKafkaTestCluster {
+
+  @ClassRule
+  public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+  private static boolean debug = true;
+
+  private static ZooKeeperLocalCluster zooKeeperLocalCluster;
+  private static KafkaLocalCluster kafkaLocalCluster;
+  private static GeodeLocalCluster geodeLocalCluster;
+
+  @BeforeClass
+  public static void setup() throws IOException, QuorumPeerConfig.ConfigException, InterruptedException {
+    startZooKeeper();
+    startKafka();
+    startGeode();
+  }
+
+  @AfterClass
+  public static void shutdown() {
+    kafkaLocalCluster.stop();
+    geodeLocalCluster.stop();
+  }
+
+  private ClientCache createGeodeClient() {
+    return new ClientCacheFactory().addPoolLocator("127.0.0.1", 10334).create();
+  }
+
+  private static void startZooKeeper() throws IOException, QuorumPeerConfig.ConfigException {
+    zooKeeperLocalCluster = new ZooKeeperLocalCluster(getZooKeeperProperties());
+    zooKeeperLocalCluster.start();
+  }
+
+  private static void startKafka() throws IOException, InterruptedException, QuorumPeerConfig.ConfigException {
+    kafkaLocalCluster = new KafkaLocalCluster(getKafkaConfig());
+    kafkaLocalCluster.start();
+  }
+
+  private static void startGeode() throws IOException, InterruptedException {
+    geodeLocalCluster = new GeodeLocalCluster();
+    geodeLocalCluster.start();
+  }
+
+  private static  Properties getZooKeeperProperties() throws IOException {
+    Properties properties = new Properties();
+    properties.setProperty("dataDir", (debug)? "/tmp/zookeeper" :temporaryFolder.newFolder("zookeeper").getAbsolutePath());
+    properties.setProperty("clientPort", "2181");
+    properties.setProperty("tickTime", "2000");
+    return properties;
+  }
+
+
+  private static Properties getKafkaConfig() throws IOException {
+    int BROKER_PORT = 8888;
+    Properties props = new Properties();
+
+    props.put("broker.id", "0");
+    props.put("zookeeper.connect", "localhost:2181");
+    props.put("host.name", "localHost");
+    props.put("port", BROKER_PORT);
+    props.put("offsets.topic.replication.factor", "1");
+    props.put("log.dir", (debug)? "/tmp/kafka" : temporaryFolder.newFolder("kafka").getAbsolutePath());
+    props.put("log.flush.interval.messages", "1");
+    props.put("log.flush.interval.ms", "10");
+
+
+    //Connector configs
+    props.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSource.class.getName());
+    props.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-source-connector");
+    props.put(ConnectorConfig.TASKS_MAX_CONFIG, "1");
+
+    //Specifically GeodeKafka connector configs
+
+    /*
+    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");
+     */
+
+/*
+name=file-source
+# The class implementing the connector
+connector.class=FileStreamSource
+# Maximum number of tasks to run for this connector instance
+tasks.max=1
+# The input file (path relative to worker's working directory)
+# This is the only setting specific to the FileStreamSource
+file=test.txt
+# The output topic in Kafka
+topic=connect-test
+ */
+
+    return props;
+  }
+
+
+
+  @Test
+  public void testX() throws InterruptedException {
+    ClientCache client = createGeodeClient();
+    Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create("someRegion");
+    region.put("JASON KEY", "JASON VALUE");
+    System.out.println("TEST COMPLETE!");
+  }
+
+}
diff --git a/src/test/java/kafka/GeodeLocalCluster.java b/src/test/java/kafka/GeodeLocalCluster.java
new file mode 100644
index 0000000..fd72dec
--- /dev/null
+++ b/src/test/java/kafka/GeodeLocalCluster.java
@@ -0,0 +1,29 @@
+package kafka;
+
+import java.io.IOException;
+
+public class GeodeLocalCluster {
+
+    private JavaProcess locatorProcess;
+    private JavaProcess serverProcess;
+
+    public GeodeLocalCluster() {
+        locatorProcess = new JavaProcess(LocatorLauncherWrapper.class);
+        serverProcess = new JavaProcess(ServerLauncherWrapper.class);
+    }
+
+    public void start() throws IOException, InterruptedException {
+        locatorProcess.exec("10334");
+        Thread.sleep(30000);
+        System.out.println("is alive?" + locatorProcess.process.isAlive());
+        serverProcess.exec("40404");
+        Thread.sleep(30000);
+
+    }
+
+    public void stop() {
+        serverProcess.destroy();
+        locatorProcess.destroy();
+    }
+}
+
diff --git a/src/test/java/kafka/JavaProcess.java b/src/test/java/kafka/JavaProcess.java
new file mode 100644
index 0000000..30edfef
--- /dev/null
+++ b/src/test/java/kafka/JavaProcess.java
@@ -0,0 +1,45 @@
+package kafka;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+
+public class JavaProcess {
+
+    public Process process;
+    Class classWithMain;
+
+    public JavaProcess(Class classWithmain) {
+        this.classWithMain = classWithmain;
+    }
+
+    public void exec(String... args) throws IOException, InterruptedException {
+        String java = System.getProperty("java.home") + File.separator + "bin" + File.separator + "java";
+        String classpath = System.getProperty("java.class.path");
+        String className = classWithMain.getName();
+
+        ProcessBuilder builder = new ProcessBuilder(
+                java, "-cp", classpath, className, convertArgsToString(args));
+
+        process = builder.inheritIO().start();
+    }
+
+    private String convertArgsToString(String... args) {
+        String string = "";
+        for(String arg: args) {
+            string += arg;
+        }
+        return string;
+    }
+
+    public void waitFor() throws InterruptedException {
+        process.waitFor();
+    }
+
+    public void destroy() {
+        process.destroy();
+    }
+
+
+
+}
\ No newline at end of file
diff --git a/src/test/java/kafka/KafkaLocalCluster.java b/src/test/java/kafka/KafkaLocalCluster.java
new file mode 100644
index 0000000..cd2a3df
--- /dev/null
+++ b/src/test/java/kafka/KafkaLocalCluster.java
@@ -0,0 +1,32 @@
+package kafka;
+
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServerStartable;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public class KafkaLocalCluster {
+
+  KafkaServerStartable kafka;
+
+  public KafkaLocalCluster(Properties kafkaProperties) throws IOException, InterruptedException {
+    KafkaConfig kafkaConfig = new KafkaConfig(kafkaProperties);
+    kafka = new KafkaServerStartable(kafkaConfig);
+  }
+
+  public void start() {
+    try {
+      kafka.startup();
+      System.out.println("Kafka started up");
+    }
+    catch (Throwable t) {
+      System.out.println(t);
+    }
+  }
+
+
+  public void stop() {
+    kafka.shutdown();
+  }
+}
diff --git a/src/test/java/kafka/LocatorLauncherWrapper.java b/src/test/java/kafka/LocatorLauncherWrapper.java
new file mode 100644
index 0000000..fe351be
--- /dev/null
+++ b/src/test/java/kafka/LocatorLauncherWrapper.java
@@ -0,0 +1,35 @@
+package kafka;
+
+import org.apache.geode.distributed.ConfigurationProperties;
+import org.apache.geode.distributed.Locator;
+import org.apache.geode.distributed.LocatorLauncher;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.Properties;
+
+public class LocatorLauncherWrapper {
+
+    public static void main(String[] args) throws IOException {
+        Properties properties = new Properties();
+//        String statsFile = new File(context.getOutputDir(), "stats.gfs").getAbsolutePath();
+//        properties.setProperty(ConfigurationPropert/**/ies.STATISTIC_ARCHIVE_FILE, statsFile);
+        properties.setProperty(ConfigurationProperties.NAME, "locator1");
+        Locator.startLocatorAndDS(10334, null, properties);
+
+//
+//        LocatorLauncher locatorLauncher  = new LocatorLauncher.Builder()
+//                .setMemberName("locator1")
+////                .setPort(Integer.valueOf(args[0]))
+////                .setBindAddress("localhost")
+//                .build();
+//
+//        locatorLauncher.start();
+//        while (!locatorLauncher.isRunning()) {
+//
+//        }
+//        System.out.println(locatorLauncher.getBindAddress() + ":" + locatorLauncher.getPort());
+
+    }
+}
diff --git a/src/test/java/kafka/ServerLauncherWrapper.java b/src/test/java/kafka/ServerLauncherWrapper.java
new file mode 100644
index 0000000..68161c2
--- /dev/null
+++ b/src/test/java/kafka/ServerLauncherWrapper.java
@@ -0,0 +1,48 @@
+package kafka;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.distributed.ConfigurationProperties;
+import org.apache.geode.distributed.ServerLauncher;
+import org.apache.geode.pdx.ReflectionBasedAutoSerializer;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public class ServerLauncherWrapper {
+
+    public static void main(String... args) throws IOException {
+//        ServerLauncher serverLauncher  = new ServerLauncher.Builder()
+//                .setMemberName("server1")
+////                .setServerPort(Integer.valueOf(args[0]))
+////                .setServerBindAddress("localhost")
+//              //  .set("locators", "localhost[10334]")
+////                .set("jmx-manager", "true")
+////                .set("jmx-manager-start", "true")
+//                .build();
+//
+//        serverLauncher.start();
+//        System.out.println("Geode Server Launcher complete");
+
+
+
+
+        Properties properties = new Properties();
+        String locatorString = "localhost[10334]";
+//        String statsFile = new File(context.getOutputDir(), "stats.gfs").getAbsolutePath();
+        Cache cache = new CacheFactory(properties)
+//                .setPdxSerializer(new ReflectionBasedAutoSerializer("benchmark.geode.data.*"))
+                .set(ConfigurationProperties.LOCATORS, locatorString)
+                .set(ConfigurationProperties.NAME,
+                        "server-1")
+                .set(ConfigurationProperties.LOG_FILE, "/Users/jhuynh/Pivotal/geode-kafka-connector/server1.log")
+                .set(ConfigurationProperties.LOG_LEVEL, "info")
+//               .set(ConfigurationProperties.STATISTIC_ARCHIVE_FILE, statsFile)
+                .create();
+        CacheServer cacheServer = cache.addCacheServer();
+        cacheServer.setPort(0);
+        cacheServer.setMaxConnections(Integer.MAX_VALUE);
+        cacheServer.start();
+    }
+}
diff --git a/src/test/java/kafka/ZooKeeperLocalCluster.java b/src/test/java/kafka/ZooKeeperLocalCluster.java
new file mode 100644
index 0000000..8b23f53
--- /dev/null
+++ b/src/test/java/kafka/ZooKeeperLocalCluster.java
@@ -0,0 +1,42 @@
+package kafka;
+
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServerMain;
+import org.apache.zookeeper.server.admin.AdminServer;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public class ZooKeeperLocalCluster {
+
+  ZooKeeperServerMain zooKeeperServer;
+  private Properties zooKeeperProperties;
+  Thread zooKeeperThread;
+
+  public ZooKeeperLocalCluster(Properties zooKeeperProperties) {
+    this.zooKeeperProperties = zooKeeperProperties;
+  }
+
+  public void start() throws IOException, QuorumPeerConfig.ConfigException {
+    QuorumPeerConfig quorumConfiguration = new QuorumPeerConfig();
+    quorumConfiguration.parseProperties(zooKeeperProperties);
+
+    zooKeeperServer = new ZooKeeperServerMain();
+    final ServerConfig configuration = new ServerConfig();
+    configuration.readFrom(quorumConfiguration);
+
+    zooKeeperThread = new Thread() {
+      public void run() {
+        try {
+          zooKeeperServer.runFromConfig(configuration);
+        } catch (IOException | AdminServer.AdminServerException e) {
+          System.out.println("ZooKeeper Failed");
+          e.printStackTrace(System.err);
+        }
+      }
+    };
+    zooKeeperThread.start();
+    System.out.println("ZooKeeper thread started");
+  }
+}