WIP

Author: Don Tregonning

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

@@ -19,6 +19,8 @@
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.kafka.connect.header.Headers;
 import org.apache.kafka.connect.sink.SinkRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * SplunkSinkRecord provides helper functionality to enable Header support for the Splunk Connect for Kafka, Namely
@@ -29,6 +31,7 @@
  * @since       1.1.0
  */
 public class SplunkSinkRecord {
+    private static final Logger log = LoggerFactory.getLogger(SplunkSinkRecord.class);
     Headers headers;
     SplunkSinkConnectorConfig connectorConfig;
     String splunkHeaderIndex = "";
@@ -50,7 +53,10 @@ public SplunkSinkRecord() {}
     public SplunkSinkRecord(SinkRecord record, SplunkSinkConnectorConfig connectorConfig) {
         this.connectorConfig = connectorConfig;
         this.headers = record.headers();
-        setMetadataValues();
+        if(this.headers != null) {
+            log.info("not null headers");
+            setMetadataValues();
+        }
     }
 
     /**
@@ -77,10 +83,24 @@ protected boolean compareRecordHeaders(SinkRecord record) {
     }
 
     private void setMetadataValues() {
-        splunkHeaderIndex = this.headers.lastWithName(connectorConfig.headerIndex).value().toString();
-        splunkHeaderHost = this.headers.lastWithName(connectorConfig.headerHost).value().toString();
-        splunkHeaderSource = this.headers.lastWithName(connectorConfig.headerSource).value().toString();
-        splunkHeaderSourcetype = this.headers.lastWithName(connectorConfig.headerSourcetype).value().toString();
+        log.info("Made it to setMetadataValues");
+
+        if(this.headers.lastWithName(connectorConfig.headerIndex).value() != null) {
+            splunkHeaderIndex = this.headers.lastWithName(connectorConfig.headerIndex).value().toString();
+        }
+        if(this.headers.lastWithName(connectorConfig.headerHost).value() != null) {
+            splunkHeaderHost = this.headers.lastWithName(connectorConfig.headerHost).value().toString();
+        }
+        if(this.headers.lastWithName(connectorConfig.headerSource).value() != null) {
+            splunkHeaderSource = this.headers.lastWithName(connectorConfig.headerSource).value().toString();
+        }
+        if(this.headers.lastWithName(connectorConfig.headerSourcetype).value() != null) {
+            splunkHeaderSourcetype = this.headers.lastWithName(connectorConfig.headerSourcetype).value().toString();
+        }
+    }
+
+    public String id() {
+        return
     }
 
     @Override

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

@@ -16,6 +16,7 @@
 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;
@@ -143,7 +144,7 @@ private void preventTooManyOutstandingEvents() {
 
     private void handleRaw(final Collection<SinkRecord> records) {
         if(connectorConfig.headerSupport) {
-            handleRecordsWithHeader(records);
+            if(records != null) { handleRecordsWithHeader(records); }
         }
 
         else if (connectorConfig.hasMetaDataConfigured()) {
@@ -160,27 +161,67 @@ else if (connectorConfig.hasMetaDataConfigured()) {
     }
 
     private void handleRecordsWithHeader(final Collection<SinkRecord> records) {
-        HashMap<SplunkSinkRecord, ArrayList<SinkRecord>> recordsWithSameHeaders = new HashMap<>();
+        log.info("Inside handle records");
+
+        HashMap<String, ArrayList<SinkRecord>> recordsWithSameHeaders = new HashMap<>();
         SplunkSinkRecord splunkSinkRecord;
         for (SinkRecord record : records) {
-            splunkSinkRecord = new SplunkSinkRecord(record, connectorConfig);
+            log.info("Inside loop");
 
-            if (!recordsWithSameHeaders.containsKey(splunkSinkRecord)) {
-                recordsWithSameHeaders.put(splunkSinkRecord, new ArrayList<>());
+            // splunkSinkRecord = new SplunkSinkRecord(record, connectorConfig);
+            String key = headerId(record);
+            if (!recordsWithSameHeaders.containsKey(key)) {
+                recordsWithSameHeaders.put(key, new ArrayList<>());
             }
             ArrayList<SinkRecord> recordList = recordsWithSameHeaders.get(record);
             recordList.add(record);
-            recordsWithSameHeaders.put(splunkSinkRecord, recordList);
+            recordsWithSameHeaders.put(key, recordList);
         }
 
+        int index = 0;
         Iterator itr = recordsWithSameHeaders.entrySet().iterator();
         while(itr.hasNext()) {
+            log.info("Sending Log {}", index);
             Map.Entry set = (Map.Entry)itr.next();
             SplunkSinkRecord splunkSinkRecordKey = (SplunkSinkRecord)set.getKey();
             ArrayList<SinkRecord> recordArrayList = (ArrayList)set.getValue();
-            EventBatch batch = createRawHeaderEventBatch(1);
+            EventBatch batch = createRawHeaderEventBatch(splunkSinkRecordKey);
             sendEvents(recordArrayList, batch);
+            index++;
+        }
+    }
+
+    public String headerId(SinkRecord sinkRecord) {
+        Headers headers = sinkRecord.headers();
+        String headerId = "";
+
+        if(headers.lastWithName(connectorConfig.headerIndex).value() != null) {
+            headerId += headers.lastWithName(connectorConfig.headerIndex).value().toString();
         }
+
+        insertheaderToken(headerId);
+
+        if(headers.lastWithName(connectorConfig.headerHost).value() != null) {
+            headerId += headers.lastWithName(connectorConfig.headerHost).value().toString();
+        }
+
+        insertheaderToken(headerId);
+
+        if(headers.lastWithName(connectorConfig.headerSource).value() != null) {
+            headerId += headers.lastWithName(connectorConfig.headerSource).value().toString();
+        }
+
+        insertheaderToken(headerId);
+
+        if(headers.lastWithName(connectorConfig.headerSourcetype).value() != null) {
+            headerId += headers.lastWithName(connectorConfig.headerSourcetype).value().toString();
+        }
+
+        return headerId;
+    }
+
+    public String insertheaderToken(String id) {
+        return id += "$$$";
     }
 
     private void handleEvent(final Collection<SinkRecord> records) {

src/test/java/com/splunk/kafka/connect/SplunkSinkTaskTest.java

@@ -226,6 +226,12 @@ public void putWithRawAndAckWithoutMeta() {
         putWithSuccess(true, false);
     }
 
+    @Test
+    public void handleRecordsWithHeaders() {
+
+
+    }
+
     private void putWithSuccess(boolean raw, boolean withMeta) {
         int batchSize = 100;
         int total = 1000;