diff --git a/src/main/java/geode/kafka/GeodeConnectorConfig.java b/src/main/java/geode/kafka/GeodeConnectorConfig.java
new file mode 100644
index 0000000..9ac561f
--- /dev/null
+++ b/src/main/java/geode/kafka/GeodeConnectorConfig.java
@@ -0,0 +1,132 @@
+package geode.kafka;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class GeodeConnectorConfig {
+
+    //Geode Configuration
+    public static final String DURABLE_CLIENT_ID_PREFIX = "durableClientId";
+    public static final String DEFAULT_DURABLE_CLIENT_ID = "";
+    public static final String DURABLE_CLIENT_TIME_OUT = "durableClientTimeout";
+    public static final String DEFAULT_DURABLE_CLIENT_TIMEOUT = "60000";
+
+
+    //GeodeKafka Specific Configuration
+    public static final String TASK_ID = "GEODE_TASK_ID"; //One config per task
+
+    public static final String CQ_PREFIX = "cqPrefix";
+    public static final String DEFAULT_CQ_PREFIX = "cqForGeodeKafka";
+    /**
+     * Specifies which Locators to connect to Apache Geode
+     */
+    public static final String LOCATORS = "locators";
+    public static final String DEFAULT_LOCATOR = "localhost[10334]";
+
+    /**
+     * Specifies which Regions to connect in Apache Geode
+     */
+    public static final String REGIONS = "regions";
+
+    /**
+     * Specifies which Topics to connect in Kafka
+     */
+    public static final String TOPICS = "topics";
+
+    /**
+     * Property to describe the Source Partition in a record
+     */
+    public static final String REGION_NAME = "regionName";  //used for Source Partition Events
+
+    public static final String BATCH_SIZE = "geodeConnectorBatchSize";
+    public static final String DEFAULT_BATCH_SIZE = "100";
+
+    public static final String QUEUE_SIZE = "geodeConnectorQueueSize";
+    public static final String DEFAULT_QUEUE_SIZE = "100000";
+
+
+
+    private final int taskId;
+    private final String durableClientId;
+    private final String durableClientIdPrefix;
+    private final String durableClientTimeout;
+    private List<String> regionNames;
+    private List<String> topics;
+    private List<LocatorHostPort> locatorHostPorts;
+
+    //just for tests
+    GeodeConnectorConfig() {
+        taskId = 0;
+        durableClientId = "";
+        durableClientIdPrefix = "";
+        durableClientTimeout = "0";
+    }
+
+    public GeodeConnectorConfig(Map<String, String> connectorProperties) {
+        taskId = Integer.parseInt(connectorProperties.get(TASK_ID));
+        durableClientIdPrefix = connectorProperties.get(DURABLE_CLIENT_ID_PREFIX);
+        if (isDurable(durableClientIdPrefix)) {
+            durableClientId = durableClientIdPrefix + taskId;
+        }
+        else {
+            durableClientId = "";
+        }
+        durableClientTimeout = connectorProperties.get(DURABLE_CLIENT_TIME_OUT);
+        regionNames = parseNames(connectorProperties.get(GeodeConnectorConfig.REGIONS));
+        topics = parseNames(connectorProperties.get(GeodeConnectorConfig.TOPICS));
+        locatorHostPorts = parseLocators(connectorProperties.get(GeodeConnectorConfig.LOCATORS));
+    }
+
+    List<String> parseNames(String names) {
+        return Arrays.stream(names.split(",")).map((s) -> s.trim()).collect(Collectors.toList());
+    }
+
+    List<LocatorHostPort> parseLocators(String locators) {
+        return Arrays.stream(locators.split(",")).map((s) -> {
+            String locatorString = s.trim();
+            return parseLocator(locatorString);
+        }).collect(Collectors.toList());
+    }
+
+    private LocatorHostPort parseLocator(String locatorString) {
+        String[] splits = locatorString.split("\\[");
+        String locator = splits[0];
+        int port = Integer.parseInt(splits[1].replace("]", ""));
+        return new LocatorHostPort(locator, port);
+    }
+
+    public boolean isDurable() {
+        return isDurable(durableClientId);
+    }
+
+    /**
+     * @param durableClientId or prefix can be passed in.  Either both will be "" or both will have a value
+     * @return
+     */
+    boolean isDurable(String durableClientId) {
+        return !durableClientId.equals("");
+    }
+
+
+    public String getDurableClientId() {
+        return durableClientId;
+    }
+
+    public String getDurableClientTimeout() {
+        return durableClientTimeout;
+    }
+
+    public List<String> getRegionNames() {
+        return regionNames;
+    }
+
+    public List<String> getTopics() {
+        return topics;
+    }
+
+    public List<LocatorHostPort> getLocatorHostPorts() {
+        return locatorHostPorts;
+    }
+}
diff --git a/src/main/java/kafka/LocatorHostPort.java b/src/main/java/geode/kafka/LocatorHostPort.java
similarity index 94%
rename from src/main/java/kafka/LocatorHostPort.java
rename to src/main/java/geode/kafka/LocatorHostPort.java
index 517bad7..50d7440 100644
--- a/src/main/java/kafka/LocatorHostPort.java
+++ b/src/main/java/geode/kafka/LocatorHostPort.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 public class LocatorHostPort {
 
diff --git a/src/main/java/kafka/GeodeKafkaSink.java b/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
similarity index 98%
rename from src/main/java/kafka/GeodeKafkaSink.java
rename to src/main/java/geode/kafka/sink/GeodeKafkaSink.java
index 67a244e..68460e4 100644
--- a/src/main/java/kafka/GeodeKafkaSink.java
+++ b/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka.sink;
 
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.utils.AppInfoParser;
diff --git a/src/main/java/kafka/GeodeEvent.java b/src/main/java/geode/kafka/source/GeodeEvent.java
similarity index 94%
rename from src/main/java/kafka/GeodeEvent.java
rename to src/main/java/geode/kafka/source/GeodeEvent.java
index 805f4a0..41e37c6 100644
--- a/src/main/java/kafka/GeodeEvent.java
+++ b/src/main/java/geode/kafka/source/GeodeEvent.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka.source;
 
 import org.apache.geode.cache.query.CqEvent;
 
diff --git a/src/main/java/kafka/GeodeKafkaSource.java b/src/main/java/geode/kafka/source/GeodeKafkaSource.java
similarity index 64%
rename from src/main/java/kafka/GeodeKafkaSource.java
rename to src/main/java/geode/kafka/source/GeodeKafkaSource.java
index 95afc50..d5da62a 100644
--- a/src/main/java/kafka/GeodeKafkaSource.java
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSource.java
@@ -1,5 +1,6 @@
-package kafka;
+package geode.kafka.source;
 
+import geode.kafka.GeodeConnectorConfig;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.utils.AppInfoParser;
 import org.apache.kafka.connect.connector.Task;
@@ -10,21 +11,19 @@
 import java.util.List;
 import java.util.Map;
 
-import static kafka.GeodeConnectorConfig.BATCH_SIZE;
-import static kafka.GeodeConnectorConfig.CQ_PREFIX;
-import static kafka.GeodeConnectorConfig.DEFAULT_BATCH_SIZE;
-import static kafka.GeodeConnectorConfig.DEFAULT_CQ_PREFIX;
-import static kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_ID;
-import static kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT;
-import static kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
-import static kafka.GeodeConnectorConfig.DEFAULT_QUEUE_SIZE;
-import static kafka.GeodeConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
-import static kafka.GeodeConnectorConfig.DURABLE_CLIENT_TIME_OUT;
-import static kafka.GeodeConnectorConfig.LOCATORS;
-import static kafka.GeodeConnectorConfig.QUEUE_SIZE;
-import static kafka.GeodeConnectorConfig.REGIONS;
-import static kafka.GeodeConnectorConfig.TOPICS;
-import static kafka.GeodeKafkaSourceTask.TASK_ID;
+import static geode.kafka.GeodeConnectorConfig.BATCH_SIZE;
+import static geode.kafka.GeodeConnectorConfig.CQ_PREFIX;
+import static geode.kafka.GeodeConnectorConfig.DEFAULT_BATCH_SIZE;
+import static geode.kafka.GeodeConnectorConfig.DEFAULT_CQ_PREFIX;
+import static geode.kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_ID;
+import static geode.kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT;
+import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
+import static geode.kafka.GeodeConnectorConfig.DEFAULT_QUEUE_SIZE;
+import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
+import static geode.kafka.GeodeConnectorConfig.DURABLE_CLIENT_TIME_OUT;
+import static geode.kafka.GeodeConnectorConfig.LOCATORS;
+import static geode.kafka.GeodeConnectorConfig.QUEUE_SIZE;
+
 
 public class GeodeKafkaSource extends SourceConnector {
 
@@ -39,16 +38,14 @@
 
   @Override
   public List<Map<String, String>> taskConfigs(int maxTasks) {
-    System.out.println("GKSource: taskConfigs");
     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++) {
     //TODO partition regions and topics
-      taskProps.put(TASK_ID, "" + i);
+      taskProps.put(GeodeConnectorConfig.TASK_ID, "" + i);
       taskConfigs.add(taskProps);
     }
     return taskConfigs;
@@ -82,6 +79,7 @@
 
   @Override
   public String version() {
+    //TODO
     return AppInfoParser.getVersion();
   }
 
diff --git a/src/main/java/kafka/GeodeKafkaSourceListener.java b/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
similarity index 95%
rename from src/main/java/kafka/GeodeKafkaSourceListener.java
rename to src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
index 4c0e729..c4d6b22 100644
--- a/src/main/java/kafka/GeodeKafkaSourceListener.java
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
@@ -1,7 +1,6 @@
-package kafka;
+package geode.kafka.source;
 
 import org.apache.geode.cache.query.CqEvent;
-import org.apache.geode.cache.query.CqListener;
 import org.apache.geode.cache.query.CqStatusListener;
 
 import java.util.concurrent.BlockingQueue;
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java b/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
new file mode 100644
index 0000000..23fa141
--- /dev/null
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
@@ -0,0 +1,133 @@
+package geode.kafka.source;
+
+import geode.kafka.GeodeContext;
+import geode.kafka.GeodeConnectorConfig;
+import org.apache.geode.cache.query.CqAttributes;
+import org.apache.geode.cache.query.CqAttributesFactory;
+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.stream.Collectors;
+
+import static geode.kafka.GeodeConnectorConfig.BATCH_SIZE;
+import static geode.kafka.GeodeConnectorConfig.CQ_PREFIX;
+import static geode.kafka.GeodeConnectorConfig.QUEUE_SIZE;
+import static geode.kafka.GeodeConnectorConfig.REGION_NAME;
+
+public class GeodeKafkaSourceTask extends SourceTask {
+
+    private static final String TASK_PREFIX = "TASK";
+    private static final String DOT = ".";
+
+    //property string to pass in to identify task
+    private static final Map<String, Long> OFFSET_DEFAULT = createOffset();
+
+    private int taskId;
+    private GeodeContext geodeContext;
+    private List<String> topics;
+
+    private Map<String, Map<String, String>> sourcePartitions;
+    private static BlockingQueue<GeodeEvent> eventBuffer;
+    private int batchSize;
+
+
+    private static Map<String, Long> createOffset() {
+        Map<String, Long> offset = new HashMap<>();
+        offset.put("OFFSET", 0L);
+        return offset;
+    }
+
+    @Override
+    public String version() {
+        return null;
+    }
+
+    @Override
+    public void start(Map<String, String> props) {
+        try {
+            GeodeConnectorConfig geodeConnectorConfig = new GeodeConnectorConfig(props);
+            geodeContext = new GeodeContext(geodeConnectorConfig);
+
+            batchSize = Integer.parseInt(props.get(BATCH_SIZE));
+            int queueSize = Integer.parseInt(props.get(QUEUE_SIZE));
+            eventBuffer = new LinkedBlockingQueue<>(queueSize);
+
+            sourcePartitions = createSourcePartitionsMap(geodeConnectorConfig.getRegionNames());
+            topics = geodeConnectorConfig.getTopics();
+
+            String cqPrefix = props.get(CQ_PREFIX);
+
+            installOnGeode(geodeConnectorConfig, geodeContext, eventBuffer, cqPrefix);
+        }
+        catch (Exception e) {
+            System.out.println("Exception:" + e);
+            e.printStackTrace();
+            throw e;
+        }
+    }
+
+    @Override
+    public List<SourceRecord> poll() throws InterruptedException {
+        ArrayList<SourceRecord> records = new ArrayList<>(batchSize);
+        ArrayList<GeodeEvent> events = new ArrayList<>(batchSize);
+        if (eventBuffer.drainTo(events, batchSize) > 0) {
+            for (GeodeEvent event : events) {
+                for (String topic : topics) {
+                    records.add(new SourceRecord(sourcePartitions.get(event.getRegionName()), OFFSET_DEFAULT, topic, null, event.getEvent()));
+                }
+            }
+
+            return records;
+        }
+
+        return null;
+    }
+
+    @Override
+    public void stop() {
+        geodeContext.getClientCache().close(true);
+    }
+
+    void installOnGeode(GeodeConnectorConfig geodeConnectorConfig, GeodeContext geodeContext, BlockingQueue eventBuffer, String cqPrefix) {
+      boolean isDurable = geodeConnectorConfig.isDurable();
+        for (String region : geodeConnectorConfig.getRegionNames()) {
+            installListenersToRegion(geodeContext, taskId, eventBuffer, region, cqPrefix, isDurable);
+        }
+        if (isDurable) {
+            geodeContext.getClientCache().readyForEvents();
+        }
+    }
+
+    void installListenersToRegion(GeodeContext geodeContext, int taskId, BlockingQueue<GeodeEvent> eventBuffer, String regionName, String cqPrefix, boolean isDurable) {
+        CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
+        cqAttributesFactory.addCqListener(new GeodeKafkaSourceListener(eventBuffer, regionName));
+        CqAttributes cqAttributes = cqAttributesFactory.create();
+        geodeContext.newCq(generateCqName(taskId, cqPrefix, regionName), "select * from /" + regionName, cqAttributes,
+                isDurable);
+    }
+
+    /**
+     * converts a list of regions names into a map of source partitions
+     *
+     * @param regionNames list of regionNames
+     * @return Map<String, Map < String, String>> a map of source partitions, keyed by region name
+     */
+    Map<String, Map<String, String>> createSourcePartitionsMap(List<String> regionNames) {
+        return regionNames.stream().map(regionName -> {
+            Map<String, String> sourcePartition = new HashMap<>();
+            sourcePartition.put(REGION_NAME, regionName);
+            return sourcePartition;
+        }).collect(Collectors.toMap(s -> s.get(REGION_NAME), s -> s));
+    }
+
+    String generateCqName(int taskId, String cqPrefix, String regionName) {
+        return cqPrefix + DOT + TASK_PREFIX + taskId + DOT + regionName;
+    }
+
+}
diff --git a/src/main/java/kafka/GeodeConnectorConfig.java b/src/main/java/kafka/GeodeConnectorConfig.java
deleted file mode 100644
index 003367d..0000000
--- a/src/main/java/kafka/GeodeConnectorConfig.java
+++ /dev/null
@@ -1,41 +0,0 @@
-package kafka;
-
-public class GeodeConnectorConfig {
-
-    //Geode Configuration
-    public static final String DURABLE_CLIENT_ID_PREFIX = "durableClientId";
-    public static final String DEFAULT_DURABLE_CLIENT_ID = "";
-    public static final String DURABLE_CLIENT_TIME_OUT = "durableClientTimeout";
-    public static final String DEFAULT_DURABLE_CLIENT_TIMEOUT = "60000";
-
-
-    //GeodeKafka Specific Configuration
-    public static final String CQ_PREFIX = "cqPrefix";
-    public static final String DEFAULT_CQ_PREFIX = "cqForGeodeKafka";
-    /**
-     * Specifies which Locators to connect to Apache Geode
-     */
-    public static final String LOCATORS = "locators";
-    public static final String DEFAULT_LOCATOR = "localhost[10334]";
-
-    /**
-     * Specifies which Regions to connect in Apache Geode
-     */
-    public static final String REGIONS = "regions";
-
-    /**
-     * Specifies which Topics to connect in Kafka
-     */
-    public static final String TOPICS = "topics";
-
-    /**
-     * Property to describe the Source Partition in a record
-     */
-    public static final String REGION_NAME = "regionName";  //used for Source Partition Events
-
-    public static final String BATCH_SIZE = "geodeConnectorBatchSize";
-    public static final String DEFAULT_BATCH_SIZE = "100";
-
-    public static final String QUEUE_SIZE = "geodeConnectorQueueSize";
-    public static final String DEFAULT_QUEUE_SIZE = "100000";
-}
diff --git a/src/main/java/kafka/GeodeKafkaSourceTask.java b/src/main/java/kafka/GeodeKafkaSourceTask.java
deleted file mode 100644
index 896b602..0000000
--- a/src/main/java/kafka/GeodeKafkaSourceTask.java
+++ /dev/null
@@ -1,196 +0,0 @@
-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.CqException;
-import org.apache.geode.cache.query.CqExistsException;
-import org.apache.geode.cache.query.RegionNotFoundException;
-import org.apache.kafka.connect.source.SourceRecord;
-import org.apache.kafka.connect.source.SourceTask;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-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.stream.Collectors;
-
-import static kafka.GeodeConnectorConfig.BATCH_SIZE;
-import static kafka.GeodeConnectorConfig.CQ_PREFIX;
-import static kafka.GeodeConnectorConfig.DEFAULT_CQ_PREFIX;
-import static kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_ID;
-import static kafka.GeodeConnectorConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT;
-import static kafka.GeodeConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
-import static kafka.GeodeConnectorConfig.DURABLE_CLIENT_TIME_OUT;
-import static kafka.GeodeConnectorConfig.QUEUE_SIZE;
-import static kafka.GeodeConnectorConfig.REGION_NAME;
-
-public class GeodeKafkaSourceTask extends SourceTask {
-
-    //property string to pass in to identify task
-    public static final String TASK_ID = "GEODE_TASK_ID";
-    private static final String TASK_PREFIX = "TASK";
-    private static final String DOT = ".";
-    private static final Map<String, Long> OFFSET_DEFAULT = createOffset();
-
-    private int taskId;
-    private ClientCache clientCache;
-    private List<String> regionNames;
-    private List<String> topics;
-    private Map<String, Map<String, String>> sourcePartitions;
-    private static BlockingQueue<GeodeEvent> eventBuffer;
-    private int batchSize;
-
-
-    private static Map<String, Long> createOffset() {
-        Map<String, Long> offset = new HashMap<>();
-        offset.put("OFFSET", 0L);
-        return offset;
-    }
-
-    @Override
-    public String version() {
-        return null;
-    }
-
-    @Override
-    public void start(Map<String, String> props) {
-        try {
-            System.out.println("JASON task start");
-            taskId = Integer.parseInt(props.get(TASK_ID));
-            batchSize = Integer.parseInt(props.get(BATCH_SIZE));
-            int queueSize = Integer.parseInt(props.get(QUEUE_SIZE));
-            eventBuffer = new LinkedBlockingQueue<>(queueSize);
-
-            //grouping will be done in the source and not the task
-            regionNames = parseNames(props.get(GeodeConnectorConfig.REGIONS));
-            topics = parseNames(props.get(GeodeConnectorConfig.TOPICS));
-            sourcePartitions = createSourcePartitionsMap(regionNames);
-
-            String durableClientId = props.get(DURABLE_CLIENT_ID_PREFIX);
-            if (!durableClientId.equals("")) {
-                durableClientId += taskId;
-            }
-            System.out.println("JASON durable client id is:" + durableClientId);
-            String durableClientTimeout = props.get(DURABLE_CLIENT_TIME_OUT);
-            String cqPrefix = props.get(CQ_PREFIX);
-
-            List<LocatorHostPort> locators = parseLocators(props.get(GeodeConnectorConfig.LOCATORS));
-            installOnGeode(taskId, eventBuffer, locators, regionNames, durableClientId, durableClientTimeout, cqPrefix);
-            System.out.println("JASON task start finished");
-        }
-        catch (Exception e) {
-            System.out.println("Exception:" + e);
-            e.printStackTrace();
-            throw e;
-        }
-    }
-
-    @Override
-    public List<SourceRecord> poll() throws InterruptedException {
-        ArrayList<SourceRecord> records = new ArrayList<>(batchSize);
-        ArrayList<GeodeEvent> events = new ArrayList<>(batchSize);
-        if (eventBuffer.drainTo(events, batchSize) > 0) {
-            for (GeodeEvent event : events) {
-                for (String topic : topics) {
-                    records.add(new SourceRecord(sourcePartitions.get(event.getRegionName()), OFFSET_DEFAULT, topic, null, event.getEvent()));
-//                    records.add(new SourceRecord(sourcePartitions.get(event.getRegionName()), OFFSET_DEFAULT, topic, null, "STRING"));
-                }
-            }
-
-            return records;
-        }
-
-        return null;
-    }
-
-    @Override
-    public void stop() {
-        clientCache.close(true);
-    }
-
-    ClientCache createClientCache(List<LocatorHostPort> locators, String durableClientName, String durableClientTimeOut) {
-        ClientCacheFactory ccf = new ClientCacheFactory().set("durable-client-id", durableClientName)
-                .set("durable-client-timeout", durableClientTimeOut)
-                .setPoolSubscriptionEnabled(true);
-        for (LocatorHostPort locator: locators) {
-          ccf.addPoolLocator(locator.getHostName(), locator.getPort()).create();
-        }
-        return ccf.create();
-    }
-
-    void installOnGeode(int taskId, BlockingQueue<GeodeEvent> eventBuffer, List<LocatorHostPort> locators, List<String> regionNames, String durableClientId, String durableClientTimeout, String cqPrefix) {
-      boolean isDurable = isDurable(durableClientId);
-
-      clientCache = createClientCache(locators, durableClientId, durableClientTimeout);
-        for (String region : regionNames) {
-            installListenersToRegion(taskId, eventBuffer, region, cqPrefix, isDurable);
-        }
-        if (isDurable) {
-          clientCache.readyForEvents();
-        }
-    }
-
-    void installListenersToRegion(int taskId, BlockingQueue<GeodeEvent> eventBuffer, String regionName, String cqPrefix, boolean isDurable) {
-        CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
-        cqAttributesFactory.addCqListener(new GeodeKafkaSourceListener(eventBuffer, regionName));
-        CqAttributes cqAttributes = cqAttributesFactory.create();
-        try {
-            clientCache.getQueryService().newCq(generateCqName(taskId, cqPrefix, regionName), "select * from /" + regionName, cqAttributes,
-                    isDurable).execute();
-        } catch (CqExistsException e) {
-            System.out.println("UHH");
-            e.printStackTrace();
-        } catch (CqException | RegionNotFoundException e) {
-            System.out.println("UHH e");
-            e.printStackTrace();
-        } catch (Exception e) {
-            System.out.println("UHHHHHH " + e);
-        }
-    }
-
-
-  List<String> parseNames(String names) {
-        return Arrays.stream(names.split(",")).map((s) -> s.trim()).collect(Collectors.toList());
-    }
-
-    List<LocatorHostPort> parseLocators(String locators) {
-        return Arrays.stream(locators.split(",")).map((s) -> {
-            String locatorString = s.trim();
-            return parseLocator(locatorString);
-        }).collect(Collectors.toList());
-    }
-
-    private LocatorHostPort parseLocator(String locatorString) {
-        String[] splits = locatorString.split("\\[");
-        String locator = splits[0];
-        int port = Integer.parseInt(splits[1].replace("]", ""));
-        return new LocatorHostPort(locator, port);
-    }
-
-    boolean isDurable(String durableClientId) {
-      return !durableClientId.equals("");
-    }
-
-    String generateCqName(int taskId, String cqPrefix, String regionName) {
-      return cqPrefix + DOT + TASK_PREFIX + taskId + DOT + regionName;
-    }
-
-    /**
-     * converts a list of regions names into a map of source partitions
-     *
-     * @param regionNames list of regionNames
-     * @return Map<String, Map < String, String>> a map of source partitions, keyed by region name
-     */
-    Map<String, Map<String, String>> createSourcePartitionsMap(List<String> regionNames) {
-        return regionNames.stream().map(regionName -> {
-            Map<String, String> sourcePartition = new HashMap<>();
-            sourcePartition.put(REGION_NAME, regionName);
-            return sourcePartition;
-        }).collect(Collectors.toMap(s -> s.get(REGION_NAME), s -> s));
-    }
-}
diff --git a/src/test/java/kafka/GeodeKafkaTestCluster.java b/src/test/java/geode/kafka/GeodeKafkaTestCluster.java
similarity index 98%
rename from src/test/java/kafka/GeodeKafkaTestCluster.java
rename to src/test/java/geode/kafka/GeodeKafkaTestCluster.java
index 85c0fc4..f1a6dff 100644
--- a/src/test/java/kafka/GeodeKafkaTestCluster.java
+++ b/src/test/java/geode/kafka/GeodeKafkaTestCluster.java
@@ -1,6 +1,7 @@
-package kafka;
+package geode.kafka;
 
 import kafka.admin.RackAwareMode;
+import geode.kafka.source.GeodeKafkaSource;
 import kafka.zk.AdminZkClient;
 import kafka.zk.KafkaZkClient;
 import org.apache.geode.cache.Region;
diff --git a/src/test/java/kafka/GeodeLocalCluster.java b/src/test/java/geode/kafka/GeodeLocalCluster.java
similarity index 97%
rename from src/test/java/kafka/GeodeLocalCluster.java
rename to src/test/java/geode/kafka/GeodeLocalCluster.java
index 43ac8f5..afd3b9d 100644
--- a/src/test/java/kafka/GeodeLocalCluster.java
+++ b/src/test/java/geode/kafka/GeodeLocalCluster.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 import java.io.IOException;
 
diff --git a/src/test/java/kafka/JavaProcess.java b/src/test/java/geode/kafka/JavaProcess.java
similarity index 97%
rename from src/test/java/kafka/JavaProcess.java
rename to src/test/java/geode/kafka/JavaProcess.java
index 30edfef..fe00094 100644
--- a/src/test/java/kafka/JavaProcess.java
+++ b/src/test/java/geode/kafka/JavaProcess.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 import java.io.File;
 import java.io.IOException;
diff --git a/src/test/java/kafka/KafkaLocalCluster.java b/src/test/java/geode/kafka/KafkaLocalCluster.java
similarity index 96%
rename from src/test/java/kafka/KafkaLocalCluster.java
rename to src/test/java/geode/kafka/KafkaLocalCluster.java
index cd2a3df..57f16f4 100644
--- a/src/test/java/kafka/KafkaLocalCluster.java
+++ b/src/test/java/geode/kafka/KafkaLocalCluster.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 import kafka.server.KafkaConfig;
 import kafka.server.KafkaServerStartable;
diff --git a/src/test/java/kafka/LocatorLauncherWrapper.java b/src/test/java/geode/kafka/LocatorLauncherWrapper.java
similarity index 98%
rename from src/test/java/kafka/LocatorLauncherWrapper.java
rename to src/test/java/geode/kafka/LocatorLauncherWrapper.java
index c1a7075..57ff405 100644
--- a/src/test/java/kafka/LocatorLauncherWrapper.java
+++ b/src/test/java/geode/kafka/LocatorLauncherWrapper.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 import org.apache.geode.distributed.ConfigurationProperties;
 import org.apache.geode.distributed.Locator;
diff --git a/src/test/java/kafka/ServerLauncherWrapper.java b/src/test/java/geode/kafka/ServerLauncherWrapper.java
similarity index 98%
rename from src/test/java/kafka/ServerLauncherWrapper.java
rename to src/test/java/geode/kafka/ServerLauncherWrapper.java
index 933824e..b36a3aa 100644
--- a/src/test/java/kafka/ServerLauncherWrapper.java
+++ b/src/test/java/geode/kafka/ServerLauncherWrapper.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.CacheFactory;
diff --git a/src/test/java/kafka/WorkerAndHerderCluster.java b/src/test/java/geode/kafka/WorkerAndHerderCluster.java
similarity index 95%
rename from src/test/java/kafka/WorkerAndHerderCluster.java
rename to src/test/java/geode/kafka/WorkerAndHerderCluster.java
index 7357232..c347946 100644
--- a/src/test/java/kafka/WorkerAndHerderCluster.java
+++ b/src/test/java/geode/kafka/WorkerAndHerderCluster.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 import java.io.IOException;
 
diff --git a/src/test/java/kafka/WorkerAndHerderWrapper.java b/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
similarity index 88%
rename from src/test/java/kafka/WorkerAndHerderWrapper.java
rename to src/test/java/geode/kafka/WorkerAndHerderWrapper.java
index 5f8ccd2..cc8e27b 100644
--- a/src/test/java/kafka/WorkerAndHerderWrapper.java
+++ b/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
@@ -1,7 +1,6 @@
-package kafka;
+package geode.kafka;
 
-import org.apache.geode.distributed.ConfigurationProperties;
-import org.apache.geode.distributed.Locator;
+import geode.kafka.source.GeodeKafkaSource;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy;
 import org.apache.kafka.connect.runtime.ConnectorConfig;
@@ -14,16 +13,14 @@
 import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
 import org.apache.kafka.connect.util.ConnectUtils;
 
-import java.io.File;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Properties;
 
-import static kafka.GeodeConnectorConfig.REGIONS;
-import static kafka.GeodeConnectorConfig.TOPICS;
-import static kafka.GeodeKafkaTestCluster.TEST_REGIONS;
-import static kafka.GeodeKafkaTestCluster.TEST_TOPICS;
+import static geode.kafka.GeodeConnectorConfig.REGIONS;
+import static geode.kafka.GeodeConnectorConfig.TOPICS;
+import static geode.kafka.GeodeKafkaTestCluster.TEST_REGIONS;
+import static geode.kafka.GeodeKafkaTestCluster.TEST_TOPICS;
 
 public class WorkerAndHerderWrapper {
 
diff --git a/src/test/java/kafka/ZooKeeperLocalCluster.java b/src/test/java/geode/kafka/ZooKeeperLocalCluster.java
similarity index 98%
rename from src/test/java/kafka/ZooKeeperLocalCluster.java
rename to src/test/java/geode/kafka/ZooKeeperLocalCluster.java
index 8b23f53..a3d3433 100644
--- a/src/test/java/kafka/ZooKeeperLocalCluster.java
+++ b/src/test/java/geode/kafka/ZooKeeperLocalCluster.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka;
 
 import org.apache.zookeeper.server.ServerConfig;
 import org.apache.zookeeper.server.ZooKeeperServerMain;
diff --git a/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java b/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
new file mode 100644
index 0000000..33c260d
--- /dev/null
+++ b/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
@@ -0,0 +1,85 @@
+package geode.kafka.source;
+
+import geode.kafka.GeodeConnectorConfig;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static geode.kafka.GeodeConnectorConfig.REGION_NAME;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+public class GeodeKafkaSourceTaskTest {
+
+    @Test
+    public void cqListenerOnEventPopulatesEventsBuffer() {
+
+    }
+
+    @Test
+    public void pollReturnsEventsWhenEventBufferHasValues() {
+
+    }
+
+    @Test
+    public void regionsArePassedCorrectlyToTask() {
+
+    }
+
+    @Test
+    public void installOnGeodeShouldCallCq() {
+        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+    }
+
+
+
+
+
+    @Test
+    public void createSourcePartitionsShouldReturnAMapOfSourcePartitions() {
+        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+        List<String> regionNames = Arrays.asList(new String[]{"region1", "region2", "region3"});
+        Map<String, Map<String,String>> sourcePartitions = task.createSourcePartitionsMap(regionNames);
+        assertThat(3, is(sourcePartitions.size()));
+        assertThat(true, is(sourcePartitions.get("region1").get(REGION_NAME).equals("region1")));
+        assertThat(true, is(sourcePartitions.get("region2").get(REGION_NAME).equals("region2")));
+        assertThat(true, is(sourcePartitions.get("region3").get(REGION_NAME).equals("region3")));
+    }
+
+
+
+    @Test
+    public void listOfLocatorsShouldBeConfiguredIntoClientCache() {
+
+    }
+
+    @Test
+    public void shouldNotBeDurableIfDurableClientIdIsNull() {
+
+    }
+
+    @Test
+    public void shouldNotCallReadyForEventsIfDurableClientPrefixIsEmpty() {
+
+    }
+
+    //Source properties tests
+    @Test
+    public void propertiesShouldBeCorrectlyTranslatedToConfiguration() {
+        Map<String, String> props = new HashMap<>();
+        props.put(GeodeConnectorConfig.QUEUE_SIZE, GeodeConnectorConfig.DEFAULT_QUEUE_SIZE);
+        props.put(GeodeConnectorConfig.BATCH_SIZE, GeodeConnectorConfig.DEFAULT_BATCH_SIZE);
+
+        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+        task.start(props);
+
+//        assertThat(task.getQueueSize(GeodeConnectorConfig.QUEUE_SIZE));
+
+
+    }
+
+
+}
diff --git a/src/test/java/kafka/GeodeKafkaSourceTest.java b/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
similarity index 93%
rename from src/test/java/kafka/GeodeKafkaSourceTest.java
rename to src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
index ec6dff8..717d495 100644
--- a/src/test/java/kafka/GeodeKafkaSourceTest.java
+++ b/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
@@ -1,4 +1,4 @@
-package kafka;
+package geode.kafka.source;
 
 import org.junit.Test;
 
diff --git a/src/test/java/kafka/GeodeKafkaSourceTaskTest.java b/src/test/java/kafka/GeodeKafkaSourceTaskTest.java
deleted file mode 100644
index 2c15664..0000000
--- a/src/test/java/kafka/GeodeKafkaSourceTaskTest.java
+++ /dev/null
@@ -1,114 +0,0 @@
-package kafka;
-
-import org.junit.Test;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static kafka.GeodeConnectorConfig.REGION_NAME;
-import static org.hamcrest.CoreMatchers.allOf;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-
-public class GeodeKafkaSourceTaskTest {
-
-    @Test
-    public void cqListenerOnEventPopulatesEventsBuffer() {
-
-    }
-
-    @Test
-    public void pollReturnsEventsWhenEventBufferHasValues() {
-
-    }
-
-    @Test
-    public void regionsArePassedCorrectlyToTask() {
-
-    }
-
-    @Test
-    public void installOnGeodeShouldCallCq() {
-        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
-    }
-
-
-
-    @Test
-    public void parseRegionNamesShouldSplitOnComma() {
-        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
-        List<String> regionNames = task.parseNames("region1,region2,region3,region4");
-        assertEquals(4, regionNames.size());
-        assertThat(true, allOf(is(regionNames.contains("region1"))
-                , is(regionNames.contains("region2"))
-                , is(regionNames.contains("region3"))
-                , is(regionNames.contains("region4"))));
-    }
-
-    @Test
-    public void parseRegionNamesShouldChomp() {
-        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
-        List<String> regionNames = task.parseNames("region1, region2, region3,region4");
-        assertEquals(4, regionNames.size());
-        assertThat(true, allOf(is(regionNames instanceof List)
-                , is(regionNames.contains("region1"))
-                , is(regionNames.contains("region2"))
-                , is(regionNames.contains("region3"))
-                , is(regionNames.contains("region4"))));
-    }
-
-    @Test
-    public void createSourcePartitionsShouldReturnAMapOfSourcePartitions() {
-        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
-        List<String> regionNames = Arrays.asList(new String[]{"region1", "region2", "region3"});
-        Map<String, Map<String,String>> sourcePartitions = task.createSourcePartitionsMap(regionNames);
-        assertThat(3, is(sourcePartitions.size()));
-        assertThat(true, is(sourcePartitions.get("region1").get(REGION_NAME).equals("region1")));
-        assertThat(true, is(sourcePartitions.get("region2").get(REGION_NAME).equals("region2")));
-        assertThat(true, is(sourcePartitions.get("region3").get(REGION_NAME).equals("region3")));
-    }
-
-    @Test
-    public void shouldBeAbleToParseGeodeLocatorStrings() {
-        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
-        String locatorString="localhost[8888], localhost[8881]";
-        List<LocatorHostPort> locators = task.parseLocators(locatorString);
-        assertThat(2, is(locators.size()));
-    }
-
-    @Test
-    public void listOfLocatorsShouldBeConfiguredIntoClientCache() {
-
-    }
-
-    @Test
-    public void shouldNotBeDurableIfDurableClientIdIsNull() {
-
-    }
-
-    @Test
-    public void shouldNotCallReadyForEventsIfDurableClientPrefixIsEmpty() {
-
-    }
-
-    //Source properties tests
-    @Test
-    public void propertiesShouldBeCorrectlyTranslatedToConfiguration() {
-        Map<String, String> props = new HashMap<>();
-        props.put(GeodeConnectorConfig.QUEUE_SIZE, GeodeConnectorConfig.DEFAULT_QUEUE_SIZE);
-        props.put(GeodeConnectorConfig.BATCH_SIZE, GeodeConnectorConfig.DEFAULT_BATCH_SIZE);
-
-
-        GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
-        task.start(props);
-
-//        assertThat(task.getQueueSize(GeodeConnectorConfig.QUEUE_SIZE));
-
-
-    }
-
-
-}