Updates based on review

Author: Don Tregonning

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

@@ -61,8 +61,7 @@ public boolean equals(Object obj) {
             return new EqualsBuilder()
             .append(endpoint, other.endpoint)
             .isEquals();
-        } else {
-        return false;
         }
+        return false;
     }
 }

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

@@ -146,14 +146,16 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
             + "latency metadata will be indexed along with raw data. This setting only works in "
             + "conjunction with /event HEC endpoint (\"splunk.hec.raw\" : \"false\"). By default"
             + ", this is set to false.";
-    static final String HEC_EVENT_FORMATTED_DOC = "Ensures events already formatted correctly for HEC are parsed "
-            + "correctly in the connector, before being sent to Splunk";
+    static final String HEC_EVENT_FORMATTED_DOC = "Ensures events that are pre-formatted into the properly formatted HEC "
+            + "JSON format as per http://dev.splunk.com/view/event-collector/SP-CAAAE6P have meta-data and event data indexed "
+            + "correctly by Splunk.";
     // TBD
     static final String SSL_TRUSTSTORE_PATH_DOC = "Path on the local disk to the certificate trust store.";
     static final String SSL_TRUSTSTORE_PASSWORD_DOC = "Password for the trust store.";
 
     static final String HEADER_SUPPORT_DOC = "Setting will enable Kafka Record headers to be used for meta data override";
-    static final String HEADER_CUSTOM_DOC = "Setting will enable look for Record headers with these values and pass them ";
+    static final String HEADER_CUSTOM_DOC = "Setting will enable look for Record headers with these values and add them"
+            + "to each event if present. Custom headers are configured separated by comma for multiple headers. ex,  (\"custom_header_1,custom_header_2,custom_header_3\").";
     static final String HEADER_INDEX_DOC = "Header to use for Splunk Header Index";
     static final String HEADER_SOURCE_DOC = "Header to use for Splunk Header Source";
     static final String HEADER_SOURCETYPE_DOC = "Header to use for Splunk Header Sourcetype";

src/main/java/com/splunk/kafka/connect/KafkaHeaderUtility.java renamed to src/main/java/com/splunk/kafka/connect/SplunkSinkRecord.java

@@ -19,22 +19,22 @@
 import org.apache.kafka.connect.sink.SinkRecord;
 
 /**
- * KafkaHeaderUtility provides helper functionality to enable Header support for the Splunk Connect for Kafka, Namely
+ * SplunkSinkRecord provides helper functionality to enable Header support for the Splunk Connect for Kafka, Namely
  * Header functionality introspection and comparison.
  * <p>
  *
  * @version     1.1.0
  * @since       1.1.0
  */
-public class KafkaHeaderUtility {
+public class SplunkSinkRecord {
     Headers headers;
     SplunkSinkConnectorConfig connectorConfig;
     String splunkHeaderIndex = "";
     String splunkHeaderHost = "";
     String splunkHeaderSource = "";
     String splunkHeaderSourcetype = "";
 
-    public KafkaHeaderUtility() {}
+    public SplunkSinkRecord() {}
 
     /**
      * Creates a new Kafka Header utility object. Will take a Kafka SinkRecord and Splunk Sink Connector configuration
@@ -45,10 +45,10 @@ public KafkaHeaderUtility() {}
      * @version     1.1.0
      * @since       1.1.0
      */
-    public KafkaHeaderUtility(SinkRecord record, SplunkSinkConnectorConfig connectorConfig) {
+    public SplunkSinkRecord(SinkRecord record, SplunkSinkConnectorConfig connectorConfig) {
         this.connectorConfig = connectorConfig;
         this.headers = record.headers();
-        setMetadataValues(record.headers());
+        setMetadataValues();
     }
 
     /**
@@ -60,46 +60,39 @@ public KafkaHeaderUtility(SinkRecord record, SplunkSinkConnectorConfig connector
      * @version     1.1.0
      * @since       1.1.0
      */
-    public boolean compareRecordHeaders(SinkRecord record) {
+    protected boolean compareRecordHeaders(SinkRecord record) {
         headers = record.headers();
+        String index = headers.lastWithName(connectorConfig.headerIndex).value().toString();
+        String host = headers.lastWithName(connectorConfig.headerHost).value().toString();
+        String source = headers.lastWithName(connectorConfig.headerSource).value().toString();
+        String sourcetype = headers.lastWithName(connectorConfig.headerSourcetype).value().toString();
 
-        if(splunkHeaderIndex.equals(headers.lastWithName(connectorConfig.headerIndex).value().toString()) &&
-                splunkHeaderHost.equals(headers.lastWithName(connectorConfig.headerHost).value().toString()) &&
-                splunkHeaderSource.equals(headers.lastWithName(connectorConfig.headerSource).value().toString()) &&
-                splunkHeaderSourcetype.equals(headers.lastWithName(connectorConfig.headerSourcetype).value().toString())) {
+        if(splunkHeaderIndex.equals(index) && splunkHeaderHost.equals(host) &&
+           splunkHeaderSource.equals(source) && splunkHeaderSourcetype.equals(sourcetype))  {
             return true;
         }
         return false;
     }
 
-    public void setHeaders(Headers headers) {
-        this.headers = headers;
-        setMetadataValues(headers);
-    }
-
-    public void setMetadataValues(Headers headers) {
-        splunkHeaderIndex = headers.lastWithName(connectorConfig.headerIndex).value().toString();
-        splunkHeaderHost = headers.lastWithName(connectorConfig.headerHost).value().toString();
-        splunkHeaderSource = headers.lastWithName(connectorConfig.headerSource).value().toString();
-        splunkHeaderSourcetype = headers.lastWithName(connectorConfig.headerSourcetype).value().toString();
+    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();
     }
 
     public Headers getHeaders() {
         return headers;
     }
-
     public String getSplunkHeaderIndex() {
         return splunkHeaderIndex;
     }
-
     public String getSplunkHeaderHost() {
         return splunkHeaderHost;
     }
-
     public String getSplunkHeaderSource() {
         return splunkHeaderSource;
     }
-
     public String getSplunkHeaderSourcetype() {
         return splunkHeaderSourcetype;
     }

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

@@ -141,46 +141,40 @@ private void preventTooManyOutstandingEvents() {
 
     private void handleRaw(final Collection<SinkRecord> records) {
         if(connectorConfig.headerSupport) {
-            boolean first_record = true;
-            List<SinkRecord> likeRecords = new ArrayList<>();
-            KafkaHeaderUtility kafkaHeaderUtility = null;
-
-            for(SinkRecord record : records) {
-                if(first_record) {
-                    likeRecords.add(record);
-                    kafkaHeaderUtility = new KafkaHeaderUtility(record, connectorConfig);
-                    first_record = false;
-                    continue;
-                }
-
-                if(kafkaHeaderUtility.compareRecordHeaders(record)) {
-                    likeRecords.add(record);
-                }
-                else {
-                    EventBatch batch = createRawHeaderEventBatch(kafkaHeaderUtility);
-                    sendEvents(likeRecords, batch);
-                    likeRecords.clear();
-                    likeRecords.add(record);
-                    kafkaHeaderUtility = new KafkaHeaderUtility(record, connectorConfig);
-                }
-            }
-            EventBatch batch = createRawHeaderEventBatch(kafkaHeaderUtility);
-            sendEvents(likeRecords, batch);
+            handleRecordsWithHeader(records);
         }
+
         else if (connectorConfig.hasMetaDataConfigured()) {
             // when setup metadata - index, source, sourcetype, we need partition records for /raw
             Map<TopicPartition, Collection<SinkRecord>> partitionedRecords = partitionRecords(records);
             for (Map.Entry<TopicPartition, Collection<SinkRecord>> entry: partitionedRecords.entrySet()) {
                 EventBatch batch = createRawEventBatch(entry.getKey());
                 sendEvents(entry.getValue(), batch);
             }
-        }
-        else {
+        } else {
             EventBatch batch = createRawEventBatch(null);
             sendEvents(records, batch);
         }
     }
 
+    private void handleRecordsWithHeader(final Collection<SinkRecord> records) {
+        List<SinkRecord> recordsWithSameHeaders = new ArrayList<>();
+        SplunkSinkRecord splunkSinkRecord = new SplunkSinkRecord();
+
+        for (SinkRecord record : records) {
+            if (splunkSinkRecord.compareRecordHeaders(record)) {
+                recordsWithSameHeaders.add(record);
+                continue;
+            }
+
+            EventBatch batch = createRawHeaderEventBatch(splunkSinkRecord);
+            sendEvents(recordsWithSameHeaders, batch);
+            recordsWithSameHeaders.clear();
+            recordsWithSameHeaders.add(record);
+            splunkSinkRecord = new SplunkSinkRecord(record, connectorConfig);
+        }
+    }
+
     private void handleEvent(final Collection<SinkRecord> records) {
         EventBatch batch = new JsonEventBatch();
         sendEvents(records, batch);
@@ -223,12 +217,12 @@ private void send(final EventBatch batch) {
         }
     }
 
-    private EventBatch createRawHeaderEventBatch(KafkaHeaderUtility kafkaHeaderUtility) {
+    private EventBatch createRawHeaderEventBatch(SplunkSinkRecord splunkSinkRecord) {
         return RawEventBatch.factory()
-                .setIndex(kafkaHeaderUtility.getSplunkHeaderIndex())
-                .setSourcetype(kafkaHeaderUtility.getSplunkHeaderSourcetype())
-                .setSource(kafkaHeaderUtility.getSplunkHeaderSource())
-                .setHost(kafkaHeaderUtility.getSplunkHeaderHost())
+                .setIndex(splunkSinkRecord.getSplunkHeaderIndex())
+                .setSourcetype(splunkSinkRecord.getSplunkHeaderSourcetype())
+                .setSource(splunkSinkRecord.getSplunkHeaderSource())
+                .setHost(splunkSinkRecord.getSplunkHeaderHost())
                 .build();
 
     }
@@ -288,7 +282,7 @@ private Event createHecEventFrom(final SinkRecord record) {
         if (connectorConfig.raw) {
             RawEvent event = new RawEvent(record.value(), record);
             event.setLineBreaker(connectorConfig.lineBreaker);
-            if(connectorConfig.headerSupport) { event = (RawEvent)addHeadersToHecEvent(event, record); }
+            if(connectorConfig.headerSupport) { event = (RawEvent)addHeaders(event, record); }
             return event;
         }
 
@@ -303,13 +297,12 @@ private Event createHecEventFrom(final SinkRecord record) {
                 log.error("event does not follow correct HEC pre-formatted format", record.toString());
                 event = createHECEventNonFormatted(record);
             }
-        }
-        else {
+        } else {
             event = createHECEventNonFormatted(record);
         }
 
         if(connectorConfig.headerSupport) {
-            addHeadersToHecEvent(event, record);
+            addHeaders(event, record);
         }
 
         if (connectorConfig.trackData) {
@@ -325,41 +318,43 @@ private Event createHecEventFrom(final SinkRecord record) {
         return event;
     }
 
-    private Event addHeadersToHecEvent(Event event, SinkRecord record) {
-        //TODO: Will need exception handling around String serialization
+    private Event addHeaders(Event event, SinkRecord record) {
         Headers headers = record.headers();
-        log.debug(headers.toString());
-        if (!headers.isEmpty()) {
-            if (headers.lastWithName(connectorConfig.headerIndex) != null) {
-                log.debug("splunk_index header detected, value is: " + headers.lastWithName(connectorConfig.headerIndex).value().toString());
-                event.setIndex(headers.lastWithName(connectorConfig.headerIndex).value().toString());
-            }
-            if (headers.lastWithName(connectorConfig.headerHost) != null) {
-                log.debug("splunk_host header detected, value is: " + headers.lastWithName(connectorConfig.headerHost).value().toString());
-                event.setHost(headers.lastWithName(connectorConfig.headerHost).value().toString());
-            }
-            if (headers.lastWithName(connectorConfig.headerSource) != null) {
-                log.debug("splunk_source header detected, value is: " + headers.lastWithName(connectorConfig.headerSource).value().toString());
-                event.setSource(headers.lastWithName(connectorConfig.headerSource).value().toString());
-            }
-            if (headers.lastWithName(connectorConfig.headerSourcetype) != null) {
-                log.debug("splunk_sourcetype header detected, value is: " + headers.lastWithName(connectorConfig.headerSourcetype).value().toString());
-                event.setSourcetype(headers.lastWithName(connectorConfig.headerSourcetype).value().toString());
-            }
+        if(headers.isEmpty() &&  connectorConfig.headerCustom.isEmpty()) {
+            return event;
+        }
+
+        if (headers.lastWithName(connectorConfig.headerIndex) != null) {
+        //log.debug("splunk_index header detected, value is: " + headers.lastWithName(connectorConfig.headerIndex).value().toString());
+            event.setIndex(headers.lastWithName(connectorConfig.headerIndex).value().toString());
+        }
+        if (headers.lastWithName(connectorConfig.headerHost) != null) {
+            //log.debug("splunk_host header detected, value is: " + headers.lastWithName(connectorConfig.headerHost).value().toString());
+            event.setHost(headers.lastWithName(connectorConfig.headerHost).value().toString());
+        }
+        if (headers.lastWithName(connectorConfig.headerSource) != null) {
+            //log.debug("splunk_source header detected, value is: " + headers.lastWithName(connectorConfig.headerSource).value().toString());
+            event.setSource(headers.lastWithName(connectorConfig.headerSource).value().toString());
+        }
+        if (headers.lastWithName(connectorConfig.headerSourcetype) != null) {
+            //log.debug("splunk_sourcetype header detected, value is: " + headers.lastWithName(connectorConfig.headerSourcetype).value().toString());
+            event.setSourcetype(headers.lastWithName(connectorConfig.headerSourcetype).value().toString());
+        }
 
-            if (!connectorConfig.headerCustom.isEmpty()) {
-                String[] customHeaders = connectorConfig.headerCustom.split(",");
-                Map<String, String> headerMap = new HashMap<>();
-                for (String header : customHeaders) {
-                    if (headers.lastWithName(header) != null) {
-                        log.debug(header + " header detected, value is: " + headers.lastWithName(header).value().toString());
-                        headerMap.put(header, headers.lastWithName(header).value().toString());
-                    } else {
-                        log.debug(header + " header value not present in event.");
-                    }
+        // Custom headers are configured with a comma separated list passed in configuration
+        // "custom_header_1,custom_header_2,custom_header_3"
+        if (!connectorConfig.headerCustom.isEmpty()) {
+            String[] customHeaders = connectorConfig.headerCustom.split(",");
+            Map<String, String> headerMap = new HashMap<>();
+            for (String header : customHeaders) {
+                if (headers.lastWithName(header) != null) {
+                    log.debug(header + " header detected, value is: " + headers.lastWithName(header).value().toString());
+                    headerMap.put(header, headers.lastWithName(header).value().toString());
+                } else {
+                    log.debug(header + " header value not present in event.");
                 }
-                event.addFields(headerMap);
             }
+            event.addFields(headerMap);
         }
         return event;
     }

src/test/java/com/splunk/kafka/connect/KafkaHeaderUtilityTest.java renamed to src/test/java/com/splunk/kafka/connect/SplunkSinkRecordTest.java

@@ -17,14 +17,13 @@
 
 import org.apache.kafka.connect.header.Headers;
 import org.apache.kafka.connect.data.Schema;
-import org.apache.kafka.connect.sink.SinkConnector;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.junit.Assert;
 import org.junit.jupiter.api.Test;
 
 import java.util.Map;
 
-public class KafkaHeaderUtilityTest {
+public class SplunkSinkRecordTest {
 
     @Test
     public void checkKafkaHeaderUtilityGetters() {
@@ -43,12 +42,12 @@ public void checkKafkaHeaderUtilityGetters() {
 
         System.out.println(headers.toString());
 
-        KafkaHeaderUtility kafkaHeaderUtility = new KafkaHeaderUtility(record, connectorConfig);
+        SplunkSinkRecord splunkSinkRecord = new SplunkSinkRecord(record, connectorConfig);
 
-        Assert.assertEquals("splunk.header.index", (kafkaHeaderUtility.getSplunkHeaderIndex()));
-        Assert.assertEquals("splunk.header.host", (kafkaHeaderUtility.getSplunkHeaderHost()));
-        Assert.assertEquals("splunk.header.source", (kafkaHeaderUtility.getSplunkHeaderSource()));
-        Assert.assertEquals("splunk.header.sourcetype", (kafkaHeaderUtility.getSplunkHeaderSourcetype()));
+        Assert.assertEquals("splunk.header.index", (splunkSinkRecord.getSplunkHeaderIndex()));
+        Assert.assertEquals("splunk.header.host", (splunkSinkRecord.getSplunkHeaderHost()));
+        Assert.assertEquals("splunk.header.source", (splunkSinkRecord.getSplunkHeaderSource()));
+        Assert.assertEquals("splunk.header.sourcetype", (splunkSinkRecord.getSplunkHeaderSourcetype()));
     }
 
     @Test
@@ -66,8 +65,7 @@ public void CompareRecordHeaders() {
 
         SplunkSinkConnectorConfig connectorConfig = new SplunkSinkConnectorConfig(config);
 
-        KafkaHeaderUtility kafkaHeaderUtility = new KafkaHeaderUtility(record_1, connectorConfig);
-        kafkaHeaderUtility.setHeaders(headers_1);
+        SplunkSinkRecord splunkSinkRecord = new SplunkSinkRecord(record_1, connectorConfig);
 
         SinkRecord record_2 = setupRecord();
 
@@ -77,7 +75,7 @@ public void CompareRecordHeaders() {
         headers_2.addString("splunk.header.source", "headersource");
         headers_2.addString("splunk.header.sourcetype", "test message");
 
-        Assert.assertTrue(kafkaHeaderUtility.compareRecordHeaders(record_2));
+        Assert.assertTrue(splunkSinkRecord.compareRecordHeaders(record_2));
 
         SinkRecord record_3 = setupRecord();
 
@@ -87,7 +85,7 @@ public void CompareRecordHeaders() {
         headers_3.addString("splunk.header.source", "headersource");
         headers_3.addString("splunk.header.sourcetype", "test message");
 
-        Assert.assertFalse(kafkaHeaderUtility.compareRecordHeaders(record_3));
+        Assert.assertFalse(splunkSinkRecord.compareRecordHeaders(record_3));
     }
 
     public SinkRecord setupRecord() {