diff --git a/build.gradle b/build.gradle
index 9e23d87..c9500b0 100644
--- a/build.gradle
+++ b/build.gradle
@@ -14,6 +14,7 @@
 dependencies {
 
     compile 'org.apache.geode:geode-core:1.11.0'
+    compile 'org.apache.geode:geode-cq:1.11.0'
     compile(group: 'org.apache.kafka', name: 'connect-api', version: '2.3.1')
     compile group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.13.0'
     compile group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.13.0'
diff --git a/src/main/java/kafka/GeodeKafkaSourceTask.java b/src/main/java/kafka/GeodeKafkaSourceTask.java
index 1f74700..9608f07 100644
--- a/src/main/java/kafka/GeodeKafkaSourceTask.java
+++ b/src/main/java/kafka/GeodeKafkaSourceTask.java
@@ -8,6 +8,7 @@
 import org.apache.geode.cache.query.CqException;
 import org.apache.geode.cache.query.CqExistsException;
 import org.apache.geode.cache.query.CqListener;
+import org.apache.geode.cache.query.RegionNotFoundException;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
 
@@ -38,11 +39,12 @@
 
   @Override
   public void start(Map<String, String> props) {
+    System.out.println("JASON task start");
     batchSize = 100;
     queueSize = 100000;
     String regionName = "someRegion";
     eventBuffer = new LinkedBlockingQueue<>(queueSize);
-    topics = new String[] {"myTopic"};
+    topics = new String[] {"someTopic"};
     sourcePartition = new HashMap<>();
     sourcePartition.put(REGION_NAME, regionName);
 
@@ -50,10 +52,12 @@
     offset.put("OFFSET", 0L);
 
     installOnGeode("localHost", 10334, "someRegion");
+    System.out.println("JASON task start end");
   }
 
   @Override
   public List<SourceRecord> poll() throws InterruptedException {
+//    System.out.println("JASON polling");
     ArrayList<SourceRecord> records = new ArrayList<>(batchSize);
     ArrayList<CqEvent> events = new ArrayList<>(batchSize);
     if (eventBuffer.drainTo(events, batchSize) > 0) {
@@ -63,9 +67,11 @@
           records.add(new SourceRecord(sourcePartition, offset, topic, null, event));
       }
 
+      System.out.println("JASON we polled and returning records" + records.size());
       return records;
     }
 
+//    System.out.println("JASON we didn't poll any records");
     return null;
   }
 
@@ -80,16 +86,26 @@
         .setPoolSubscriptionEnabled(true).addPoolLocator(locatorHost, locatorPort).create();
     CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
     cqAttributesFactory.addCqListener(new GeodeKafkaSourceListener());
+    System.out.println("JASON installing on Geode");
     CqAttributes cqAttributes = cqAttributesFactory.create();
     try {
+      System.out.println("JASON installing new cq");
       clientCache.getQueryService().newCq("kafkaCQFor" + regionName, "select * from /" + regionName, cqAttributes,
-          true);
+          true).execute();
+      System.out.println("JASON finished installing cq");
     } catch (CqExistsException e) {
+      System.out.println("UHH");
       e.printStackTrace();
-    } catch (CqException e) {
+    } catch (CqException | RegionNotFoundException e) {
+      System.out.println("UHH e");
       e.printStackTrace();
     }
+    catch (Exception e) {
+      System.out.println("UHHHHHH " + e);
+    }
+    System.out.println("JASON task calling ready for events");
     clientCache.readyForEvents();
+    System.out.println("JASON task ready for events");
   }
 
   private static class GeodeKafkaSourceListener implements CqListener {
@@ -97,6 +113,7 @@
     @Override
     public void onEvent(CqEvent aCqEvent) {
       try {
+        System.out.println("JASON cqEvent and putting into eventBuffer");
         eventBuffer.offer(aCqEvent, 2, TimeUnit.SECONDS);
       } catch (InterruptedException e) {
 
diff --git a/src/test/java/kafka/GeodeKafkaTestCluster.java b/src/test/java/kafka/GeodeKafkaTestCluster.java
index cb5f9ad..2b0293f 100644
--- a/src/test/java/kafka/GeodeKafkaTestCluster.java
+++ b/src/test/java/kafka/GeodeKafkaTestCluster.java
@@ -10,6 +10,12 @@
 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.clients.consumer.Consumer;
+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.serialization.StringDeserializer;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy;
@@ -32,7 +38,10 @@
 import org.junit.rules.TemporaryFolder;
 
 import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Properties;
 
@@ -45,6 +54,8 @@
   private static ZooKeeperLocalCluster zooKeeperLocalCluster;
   private static KafkaLocalCluster kafkaLocalCluster;
   private static GeodeLocalCluster geodeLocalCluster;
+  private static WorkerAndHerderCluster workerAndHerderCluster;
+  private static Consumer<String, String> consumer;
 
   @BeforeClass
   public static void setup() throws IOException, QuorumPeerConfig.ConfigException, InterruptedException {
@@ -52,63 +63,30 @@
     startKafka();
     startGeode();
     createTopic();
+    Thread.sleep(5000);
     startWorker();
+    consumer = createConsumer();
+    Thread.sleep(5000);
   }
 
   @AfterClass
   public static void shutdown() {
+    workerAndHerderCluster.stop();
     KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181",false,200000,
             15000,10, Time.SYSTEM, "myGroup","myMetricType", null);
+
     AdminZkClient adminZkClient = new AdminZkClient(zkClient);
     adminZkClient.deleteTopic("someTopic");
+
     kafkaLocalCluster.stop();
     geodeLocalCluster.stop();
   }
 
 
-  private static void startWorker() {
-    Map props = new HashMap();
-    props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:8888");
-    props.put("offset.storage.file.filename", "/tmp/connect.offsets");
-    // fast flushing for testing.
-    props.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "10");
-
-
-    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");
-
-
-    WorkerConfig workerCfg = new StandaloneConfig(props);
-
-    MemoryOffsetBackingStore offBackingStore = new MemoryOffsetBackingStore();
-    offBackingStore.configure(workerCfg);
-
-    Worker worker = new Worker("WORKER_ID", new SystemTime(), new Plugins(props), workerCfg, offBackingStore, new AllConnectorClientConfigOverridePolicy());
-    worker.start();
-
-    Herder herder = new StandaloneHerder(worker, ConnectUtils.lookupKafkaClusterId(workerCfg), new AllConnectorClientConfigOverridePolicy());
-    herder.start();
-
-
-
-
-    Map<String, String> sourceProps = new HashMap<>();
-    sourceProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSource.class.getName());
-    sourceProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-source-connector");
-    sourceProps.put(ConnectorConfig.TASKS_MAX_CONFIG, "1");
-//
-    herder.putConnectorConfig(
-            sourceProps.get(ConnectorConfig.NAME_CONFIG),
-            sourceProps, true, (error, result)->{
-              System.out.println("CALLBACK: " + result);
-            });
-    System.out.println("Worker and herder started");
+  private static void startWorker() throws IOException, InterruptedException {
+    workerAndHerderCluster = new WorkerAndHerderCluster();
+    workerAndHerderCluster.start();
+    Thread.sleep(20000);
   }
 
   private static void createTopic() {
@@ -186,13 +164,40 @@
   }
 
 
+  //consumer props, less important, just for testing?
+  public static Consumer<String,String> createConsumer() {
+      final Properties props = new Properties();
+    props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:8888");
+    props.put(ConsumerConfig.GROUP_ID_CONFIG,
+              "myGroup");
+      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+              StringDeserializer.class.getName());
+      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+              StringDeserializer.class.getName());
+      // Create the consumer using props.
+      final Consumer<String, String> consumer =
+              new KafkaConsumer<>(props);
+      // Subscribe to the topic.
+      consumer.subscribe(Collections.singletonList("someTopic"));
+      return consumer;
+  }
 
   @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("PUT COMPLETE!");
+    region.get("JASON KEY");
+//    client.close();
+
+
+    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));
+    for (ConsumerRecord<String, String> record: records) {
+      System.out.println("JASON we consumed a record:" + record);
+    }
     System.out.println("TEST COMPLETE!");
+
   }
 
 }
diff --git a/src/test/java/kafka/GeodeLocalCluster.java b/src/test/java/kafka/GeodeLocalCluster.java
index fd72dec..43ac8f5 100644
--- a/src/test/java/kafka/GeodeLocalCluster.java
+++ b/src/test/java/kafka/GeodeLocalCluster.java
@@ -13,12 +13,12 @@
     }
 
     public void start() throws IOException, InterruptedException {
+        System.out.println("starting locator");
         locatorProcess.exec("10334");
-        Thread.sleep(30000);
+        Thread.sleep(15000);
         System.out.println("is alive?" + locatorProcess.process.isAlive());
         serverProcess.exec("40404");
         Thread.sleep(30000);
-
     }
 
     public void stop() {