Merge pull request #414 from javabrett/fix-header-npe

Fix potential NullPointerException for Kafka headers with null value

Author: VihasMakwana

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

@@ -222,31 +222,31 @@ public String headerId(SinkRecord sinkRecord) {
 
         StringBuilder headerString = new StringBuilder();
 
-        if(indexHeader != null) {
+        if(indexHeader != null && indexHeader.value() != null) {
             headerString.append(indexHeader.value().toString());
         } else if (metas != null) {
             headerString.append(metas.get("index"));
         }
 
         headerString.append(insertHeaderToken());
 
-        if(hostHeader != null) {
+        if(hostHeader != null && hostHeader.value() != null) {
             headerString.append(hostHeader.value().toString());
         } else if (metas != null) {
             headerString.append("default-host");
         }
 
         headerString.append(insertHeaderToken());
 
-        if(sourceHeader != null) {
+        if(sourceHeader != null && sourceHeader.value() != null) {
             headerString.append(sourceHeader.value().toString());
         } else if (metas != null) {
             headerString.append(metas.get("source"));
         }
 
         headerString.append(insertHeaderToken());
 
-        if(sourcetypeHeader != null) {
+        if(sourcetypeHeader != null && sourcetypeHeader.value() != null) {
             headerString.append(sourcetypeHeader.value().toString());
         } else if (metas != null) {
             headerString.append(metas.get("sourcetype"));
@@ -442,16 +442,16 @@ private Event addHeaders(Event event, SinkRecord record) {
         Header headerSource = headers.lastWithName(connectorConfig.headerSource);
         Header headerSourcetype = headers.lastWithName(connectorConfig.headerSourcetype);
 
-        if (headerIndex != null) {
+        if (headerIndex != null && headerIndex.value() != null) {
             event.setIndex(headerIndex.value().toString());
         }
-        if (headerHost != null) {
+        if (headerHost != null && headerHost.value() != null) {
             event.setHost(headerHost.value().toString());
         }
-        if (headerSource != null) {
+        if (headerSource != null && headerSource.value() != null) {
             event.setSource(headerSource.value().toString());
         }
-        if (headerSourcetype != null) {
+        if (headerSourcetype != null && headerSourcetype.value() != null) {
             event.setSourcetype(headerSourcetype.value().toString());
         }
 

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

@@ -17,19 +17,17 @@
 
 import com.splunk.hecclient.Event;
 import com.splunk.hecclient.EventBatch;
-import com.splunk.hecclient.JsonEvent;
 import com.splunk.hecclient.RawEventBatch;
-import org.apache.commons.logging.Log;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.connect.errors.RetriableException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.junit.Assert;
 import org.junit.Test;
 
-import java.text.ParseException;
 import java.util.*;
 
 public class SplunkSinkTaskTest {
@@ -242,6 +240,25 @@ public void putWithNullEvent() {
         task.stop();
     }
 
+    @Test
+    public void putWithNullIndexHeaderValue() {
+        UnitUtil uu = new UnitUtil(0);
+        Map<String, String> config = uu.createTaskConfig();
+        config.put(SplunkSinkConnectorConfig.RAW_CONF, String.valueOf(true));
+        config.put(SplunkSinkConnectorConfig.ACK_CONF, String.valueOf(true));
+        config.put(SplunkSinkConnectorConfig.MAX_BATCH_SIZE_CONF, String.valueOf(1));
+        config.put(SplunkSinkConnectorConfig.HEADER_SUPPORT_CONF, String.valueOf("true"));
+        config.put(SplunkSinkConnectorConfig.HEADER_INDEX_CONF, "index");
+        SplunkSinkTask task = new SplunkSinkTask();
+        HecMock hec = new HecMock(task);
+        hec.setSendReturnResult(HecMock.success);
+        task.setHec(hec);
+        task.start(config);
+        task.put(createSinkRecordWithNullIndexHeaderValue());
+        Assert.assertEquals(1, hec.getBatches().size());
+        task.stop();
+    }
+
     @Test
     public void putWithRawAndAck() {
         putWithSuccess(true, true);
@@ -517,6 +534,12 @@ private Collection<SinkRecord> createNullSinkRecord() {
         return records;
     }
 
+    private Collection<SinkRecord> createSinkRecordWithNullIndexHeaderValue() {
+        List<SinkRecord> records = new ArrayList<>(createSinkRecords(1));
+        records.get(0).headers().add("index", null, null);
+        return records;
+    }
+
     private List<TopicPartition> createTopicPartitionList() {
         ArrayList<TopicPartition> tps = new ArrayList<>();
         tps.add(new TopicPartition("mytopic", 1));