commit a713123163c10f4556896105c2a4b3e74eb6d4f2
author Naburun Nag <nag@cs.wisc.edu> Tue Mar 03 14:30:56 2020 -0800
committer Naburun Nag <nag@cs.wisc.edu> Tue Mar 03 14:30:56 2020 -0800
tree 4f2ff98a91c2b09b9d93142649071b6e7e32fefd
parent fffb64331a9aa78f07f53505467e08a86c6d1020

Added some logging statements.

src/main/java/org/apache/geode/kafka/sink/GeodeKafkaSinkTask.java

diff --git a/src/main/java/org/apache/geode/kafka/sink/GeodeKafkaSinkTask.java b/src/main/java/org/apache/geode/kafka/sink/GeodeKafkaSinkTask.java
index c0fde75..700688e 100644
--- a/src/main/java/org/apache/geode/kafka/sink/GeodeKafkaSinkTask.java
+++ b/src/main/java/org/apache/geode/kafka/sink/GeodeKafkaSinkTask.java
@@ -56,6 +56,7 @@
 
   @Override
   public void start(Map<String, String> props) {
+    logger.info("Starting Apache Geode sink task");
     try {
       GeodeSinkConnectorConfig geodeConnectorConfig = new GeodeSinkConnectorConfig(props);
       configure(geodeConnectorConfig);
@@ -142,6 +143,7 @@
 
   @Override
   public void stop() {
+    logger.info("Stopping task");
     geodeContext.close(false);
   }
 

src/main/java/org/apache/geode/kafka/source/GeodeKafkaSourceTask.java

diff --git a/src/main/java/org/apache/geode/kafka/source/GeodeKafkaSourceTask.java b/src/main/java/org/apache/geode/kafka/source/GeodeKafkaSourceTask.java
index a9edf95..d53a9e9 100644
--- a/src/main/java/org/apache/geode/kafka/source/GeodeKafkaSourceTask.java
+++ b/src/main/java/org/apache/geode/kafka/source/GeodeKafkaSourceTask.java
@@ -67,6 +67,7 @@
 
   @Override
   public void start(Map<String, String> props) {
+    logger.info("Starting Apache Geode source task");
     try {
       GeodeSourceConnectorConfig geodeConnectorConfig = new GeodeSourceConnectorConfig(props);
       logger.debug("GeodeKafkaSourceTask id:" + geodeConnectorConfig.getTaskId() + " starting");
@@ -89,6 +90,7 @@
       boolean loadEntireRegion = geodeConnectorConfig.getLoadEntireRegion();
       installOnGeode(geodeConnectorConfig, geodeContext, eventBufferSupplier, cqPrefix,
           loadEntireRegion);
+      logger.info("Started Apache Geode source task");
     } catch (Exception e) {
       e.printStackTrace();
       logger.error("Unable to start source task", e);
@@ -98,6 +100,7 @@
 
   @Override
   public List<SourceRecord> poll() {
+    logger.trace("Polling for new data");
     ArrayList<SourceRecord> records = new ArrayList<>(batchSize);
     ArrayList<GeodeEvent> events = new ArrayList<>(batchSize);
     if (eventBufferSupplier.get().drainTo(events, batchSize) > 0) {
@@ -117,6 +120,7 @@
 
   @Override
   public void stop() {
+    logger.info("Stopping Apache Geode source task");
     geodeContext.close(true);
   }