issue-165 - fixed categorization of like Headers

Author: Don Tregonning

src/main/java/com/splunk/hecclient/HecAckPoller.java

@@ -209,9 +209,9 @@ public void stickySessionHandler(HecChannel channel) {
             log.info("Failing {} batches for the channel {}, these will be resent by the connector.", channelBatches.size(), oldChannelId);
             if (pollerCallback != null) {
                 List<EventBatch> expired = new ArrayList<>();
-                Iterator iter = channelBatches.entrySet().iterator();
+                Iterator<Map.Entry<Long,EventBatch>> iter = channelBatches.entrySet().iterator();
                 while(iter.hasNext()) {
-                    Map.Entry<Long, EventBatch> pair = (Map.Entry) iter.next();
+                    Map.Entry<Long, EventBatch> pair = iter.next();
                     EventBatch batch = pair.getValue();
                     totalOutstandingEventBatches.decrementAndGet();
                     batch.fail();

src/main/java/com/splunk/kafka/connect/SplunkSinkRecord.java

@@ -100,7 +100,8 @@ private void setMetadataValues() {
     }
 
     public String id() {
-        return
+        return splunkHeaderIndex + "$$$" + splunkHeaderHost + "$$$"
+        + splunkHeaderSource + "$$$" + splunkHeaderSourcetype;
     }
 
     @Override

src/main/java/com/splunk/kafka/connect/SplunkSinkTask.java

@@ -16,7 +16,6 @@
 package com.splunk.kafka.connect;
 
 import com.splunk.hecclient.*;
-import com.sun.tools.corba.se.idl.StringGen;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.errors.RetriableException;
@@ -30,8 +29,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.net.ssl.SSLEngineResult;
-
 public final class SplunkSinkTask extends SinkTask implements PollerCallback {
     private static final Logger log = LoggerFactory.getLogger(SplunkSinkTask.class);
     private static final long flushWindow = 30 * 1000; // 30 seconds
@@ -161,67 +158,63 @@ else if (connectorConfig.hasMetaDataConfigured()) {
     }
 
     private void handleRecordsWithHeader(final Collection<SinkRecord> records) {
-        log.info("Inside handle records");
-
         HashMap<String, ArrayList<SinkRecord>> recordsWithSameHeaders = new HashMap<>();
-        SplunkSinkRecord splunkSinkRecord;
-        for (SinkRecord record : records) {
-            log.info("Inside loop");
 
-            // splunkSinkRecord = new SplunkSinkRecord(record, connectorConfig);
+        for (SinkRecord record : records) {
             String key = headerId(record);
             if (!recordsWithSameHeaders.containsKey(key)) {
-                recordsWithSameHeaders.put(key, new ArrayList<>());
+                ArrayList<SinkRecord> recordList = new ArrayList<SinkRecord>();
+                recordsWithSameHeaders.put(key, recordList);
             }
-            ArrayList<SinkRecord> recordList = recordsWithSameHeaders.get(record);
+            ArrayList<SinkRecord> recordList = recordsWithSameHeaders.get(key);
             recordList.add(record);
             recordsWithSameHeaders.put(key, recordList);
         }
 
         int index = 0;
-        Iterator itr = recordsWithSameHeaders.entrySet().iterator();
+        Iterator<Map.Entry<String, ArrayList<SinkRecord>>> itr = recordsWithSameHeaders.entrySet().iterator();
         while(itr.hasNext()) {
-            log.info("Sending Log {}", index);
-            Map.Entry set = (Map.Entry)itr.next();
-            SplunkSinkRecord splunkSinkRecordKey = (SplunkSinkRecord)set.getKey();
+            Map.Entry set = itr.next();
+            String splunkSinkRecordKey = (String)set.getKey();
             ArrayList<SinkRecord> recordArrayList = (ArrayList)set.getValue();
             EventBatch batch = createRawHeaderEventBatch(splunkSinkRecordKey);
             sendEvents(recordArrayList, batch);
             index++;
         }
+        log.debug("{} records have been bucketed in to {} batches",records.size(), index);
     }
 
     public String headerId(SinkRecord sinkRecord) {
         Headers headers = sinkRecord.headers();
         String headerId = "";
 
-        if(headers.lastWithName(connectorConfig.headerIndex).value() != null) {
+        if(headers.lastWithName(connectorConfig.headerIndex) != null) {
             headerId += headers.lastWithName(connectorConfig.headerIndex).value().toString();
         }
 
-        insertheaderToken(headerId);
+        headerId = insertheaderToken(headerId);
 
-        if(headers.lastWithName(connectorConfig.headerHost).value() != null) {
+        if(headers.lastWithName(connectorConfig.headerHost) != null) {
             headerId += headers.lastWithName(connectorConfig.headerHost).value().toString();
         }
 
-        insertheaderToken(headerId);
+        headerId = insertheaderToken(headerId);
 
-        if(headers.lastWithName(connectorConfig.headerSource).value() != null) {
+        if(headers.lastWithName(connectorConfig.headerSource) != null) {
             headerId += headers.lastWithName(connectorConfig.headerSource).value().toString();
         }
 
-        insertheaderToken(headerId);
+        headerId = insertheaderToken(headerId);
 
-        if(headers.lastWithName(connectorConfig.headerSourcetype).value() != null) {
+        if(headers.lastWithName(connectorConfig.headerSourcetype) != null) {
             headerId += headers.lastWithName(connectorConfig.headerSourcetype).value().toString();
         }
 
         return headerId;
     }
 
     public String insertheaderToken(String id) {
-        return id += "$$$";
+        return id + "$$$";
     }
 
     private void handleEvent(final Collection<SinkRecord> records) {
@@ -266,15 +259,17 @@ private void send(final EventBatch batch) {
         }
     }
 
-    private EventBatch createRawHeaderEventBatch(SplunkSinkRecord splunkSinkRecord) {
+    private EventBatch createRawHeaderEventBatch(String splunkSinkRecord) {
+        String[] split = splunkSinkRecord.split("[$]{3}");
+
         return RawEventBatch.factory()
-                .setIndex(splunkSinkRecord.getSplunkHeaderIndex())
-                .setSourcetype(splunkSinkRecord.getSplunkHeaderSourcetype())
-                .setSource(splunkSinkRecord.getSplunkHeaderSource())
-                .setHost(splunkSinkRecord.getSplunkHeaderHost())
+                .setIndex(split[0])
+                .setSourcetype(split[1])
+                .setSource(split[2])
+                .setHost(split[3])
                 .build();
-
     }
+
     // setup metadata on RawEventBatch
     private EventBatch createRawEventBatch(final TopicPartition tp) {
         if (tp == null) {