Merge pull request #57 from confluentinc/CC-26777

CC-26777: Cherry pick fix for potential NullPointerException for Kafka headers with null value.

Author: 100maygupta

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

@@ -212,7 +212,7 @@ 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) {
@@ -222,7 +222,7 @@ public String headerId(SinkRecord sinkRecord) {
 
         headerString.append(insertHeaderToken());
 
-        if(hostHeader != null) {
+        if(hostHeader != null && hostHeader.value() != null) {
             headerString.append(hostHeader.value().toString());
         } else {
             if(metas != null) {
@@ -232,7 +232,7 @@ public String headerId(SinkRecord sinkRecord) {
 
         headerString.append(insertHeaderToken());
 
-        if(sourceHeader != null) {
+        if(sourceHeader != null && sourceHeader.value() != null) {
             headerString.append(sourceHeader.value().toString());
         } else {
             if(metas != null) {
@@ -242,7 +242,7 @@ public String headerId(SinkRecord sinkRecord) {
 
         headerString.append(insertHeaderToken());
 
-        if(sourcetypeHeader != null) {
+        if(sourcetypeHeader != null && sourcetypeHeader.value() != null) {
             headerString.append(sourcetypeHeader.value().toString());
         } else {
             if(metas != null) {
@@ -438,16 +438,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

@@ -241,6 +241,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);
@@ -341,6 +360,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));