issue-165 Modify raw batching functionality to use hashmap instad of list

Author: Don Tregonning

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

@@ -15,6 +15,8 @@
  */
 package com.splunk.kafka.connect;
 
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.kafka.connect.header.Headers;
 import org.apache.kafka.connect.sink.SinkRecord;
 
@@ -81,6 +83,30 @@ private void setMetadataValues() {
         splunkHeaderSourcetype = this.headers.lastWithName(connectorConfig.headerSourcetype).value().toString();
     }
 
+    @Override
+    public int hashCode() {
+        return new HashCodeBuilder()
+                .append(splunkHeaderIndex)
+                .append(splunkHeaderHost)
+                .append(splunkHeaderSource)
+                .append(splunkHeaderSourcetype)
+                .toHashCode();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj instanceof SplunkSinkRecord) {
+            final SplunkSinkRecord other = (SplunkSinkRecord) obj;
+            return new EqualsBuilder()
+                    .append(splunkHeaderIndex, other.splunkHeaderIndex)
+                    .append(splunkHeaderHost, other.splunkHeaderHost)
+                    .append(splunkHeaderSource, other.splunkHeaderSource)
+                    .append(splunkHeaderSourcetype, other.splunkHeaderSourcetype)
+                    .isEquals();
+        }
+        return false;
+    }
+
     public Headers getHeaders() {
         return headers;
     }

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

@@ -29,6 +29,8 @@
 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
@@ -158,20 +160,26 @@ else if (connectorConfig.hasMetaDataConfigured()) {
     }
 
     private void handleRecordsWithHeader(final Collection<SinkRecord> records) {
-        List<SinkRecord> recordsWithSameHeaders = new ArrayList<>();
-        SplunkSinkRecord splunkSinkRecord = new SplunkSinkRecord();
-
+        HashMap<SplunkSinkRecord, ArrayList<SinkRecord>> recordsWithSameHeaders = new HashMap<>();
+        SplunkSinkRecord 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);
+
+            if (!recordsWithSameHeaders.containsKey(splunkSinkRecord)) {
+                recordsWithSameHeaders.put(splunkSinkRecord, new ArrayList<>());
+            }
+            ArrayList<SinkRecord> recordList = recordsWithSameHeaders.get(record);
+            recordList.add(record);
+            recordsWithSameHeaders.put(splunkSinkRecord, recordList);
+        }
+
+        Iterator itr = recordsWithSameHeaders.entrySet().iterator();
+        while(itr.hasNext()) {
+            Map.Entry set = (Map.Entry)itr.next();
+            SplunkSinkRecord splunkSinkRecordKey = (SplunkSinkRecord)set.getKey();
+            ArrayList<SinkRecord> recordArrayList = (ArrayList)set.getValue();
+            EventBatch batch = createRawHeaderEventBatch(1);
+            sendEvents(recordArrayList, batch);
         }
     }