Perf fixes (#22)

* Fixed tight spin lock at poll which can result in 100% CPU if no new messages
* Fixed leak in processor object use
* Updated travis to also publish to local gh-pages if on SolaceDev
* Fine-tuned spin turns for poll

Author: bczoma

.travis.yml

@@ -27,4 +27,20 @@ after_success:
     git remote add origin-pages https://${GH_TOKEN}@github.com/SolaceProducts/pubsubplus-connector-kafka-source.git > /dev/null 2>&1;
     git push --quiet --set-upstream origin-pages gh-pages;
     echo "Updated and pushed GH pages!";
+  elif [ "$TRAVIS_PULL_REQUEST" = "false" ] && ! [ $(echo $TRAVIS_REPO_SLUG | grep "SolaceProducts") ]; then
+    git config --global user.email "travis@travis-ci.org";
+    git config --global user.name "travis-ci";
+    mkdir gh-pages; # Now update gh-pages
+    git clone --quiet --branch=gh-pages https://${GH_TOKEN}@github.com/$TRAVIS_REPO_SLUG gh-pages > /dev/null 2>&1;
+    rm gh-pages/downloads/pubsubplus-connector-kafka-source*
+    mv build/distributions/pubsubplus-connector-kafka-source* gh-pages/downloads
+    cd gh-pages;
+    pushd downloads
+    cp index.template index.html; FILENAME=`find . | grep *.zip | cut -d'/' -f2 | sed 's/.\{4\}$//'`;  sed -i "s/CONNECTOR_NAME/$FILENAME/g" index.html;
+    popd;
+    git add -f .;
+    git commit -m "Latest connector distribution on successful travis build $TRAVIS_BUILD_NUMBER auto-pushed to gh-pages";
+    git remote add origin-pages https://${GH_TOKEN}@github.com/$TRAVIS_REPO_SLUG.git > /dev/null 2>&1;
+    git push --quiet --set-upstream origin-pages gh-pages;
+    echo "Updated and pushed GH pages!";
   fi

gradle.properties

@@ -1 +1 @@
-version=2.0.1
+version=2.0.2

src/main/java/com/solace/connector/kafka/connect/source/SolaceSourceTask.java

@@ -24,19 +24,12 @@
 import com.solacesystems.jcsmp.JCSMPException;
 import com.solacesystems.jcsmp.JCSMPProperties;
 import com.solacesystems.jcsmp.JCSMPSession;
-import com.solacesystems.jcsmp.JCSMPSessionStats;
-import com.solacesystems.jcsmp.statistics.StatType;
-
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Enumeration;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
 import org.slf4j.Logger;
@@ -59,7 +52,7 @@ public class SolaceSourceTask extends SourceTask { // implements XMLMessageListe
   String skafkaTopic;
   SolaceSourceTopicListener topicListener = null;
   SolaceSourceQueueConsumer queueConsumer = null;
-  
+  private int spinTurns = 0;
   private volatile boolean shuttingDown = false;
 
   // private Class<?> cProcessor;
@@ -74,6 +67,16 @@ public String version() {
   public void start(Map<String, String> props) {
 
     connectorConfig = new SolaceSourceConnectorConfig(props);
+    try {
+      processor = connectorConfig
+          .getConfiguredInstance(SolaceSourceConstants
+              .SOL_MESSAGE_PROCESSOR, SolMessageProcessorIF.class);
+    } catch (Exception e) {
+      log.info(
+          "================ Encountered exception in creating the message processor."
+          + " Cause: {}, Stacktrace: {} ",
+          e.getCause(), e.getStackTrace());
+    }
     skafkaTopic = connectorConfig.getString(SolaceSourceConstants.KAFKA_TOPIC);
     solSessionHandler = new SolSessionHandler(connectorConfig);
     try {
@@ -106,20 +109,23 @@ public void start(Map<String, String> props) {
   public synchronized List<SourceRecord> poll() throws InterruptedException {
 
     if (shuttingDown || ingressMessages.size() == 0) {
-        return null;  // Nothing to do, return control
+      spinTurns++;
+      if (spinTurns > 100) {
+        spinTurns = 0;
+        Thread.sleep(1);
+      }
+      return null;  // Nothing to do, return control
     }
     // There is at least one message to process
+    spinTurns = 0; // init spinTurns again
     List<SourceRecord> records = new ArrayList<>();
     int processedInIhisBatch = 0;
     int count = 0;
     int arraySize = ingressMessages.size();
     while (count < arraySize) {
       BytesXMLMessage msg = ingressMessages.take();
       try {
-        processor = connectorConfig
-            .getConfiguredInstance(SolaceSourceConstants
-                .SOL_MESSAGE_PROCESSOR, SolMessageProcessorIF.class)
-            .process(connectorConfig.getString(SolaceSourceConstants.SOL_KAFKA_MESSAGE_KEY), msg);
+          processor.process(connectorConfig.getString(SolaceSourceConstants.SOL_KAFKA_MESSAGE_KEY), msg);
       } catch (Exception e) {
         log.info(
             "================ Encountered exception in message processing....discarded."

src/main/java/com/solace/connector/kafka/connect/source/VersionUtil.java

@@ -7,7 +7,7 @@ public class VersionUtil {
    */
   public static String getVersion() {
 
-    return "2.0.1";
+    return "2.0.2";
 
   }