Merge pull request #319 from splunk/line-breaker_fix

Fix special character problem in line-breaker rule

Author: chaitanyaphalak

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

@@ -252,7 +252,11 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
         useRecordTimestamp = getBoolean(USE_RECORD_TIMESTAMP_CONF);
         maxBatchSize = getInt(MAX_BATCH_SIZE_CONF);
         numberOfThreads = getInt(HEC_THREDS_CONF);
-        lineBreaker = getString(LINE_BREAKER_CONF);
+        if (taskConfig.get(LINE_BREAKER_CONF) != null && taskConfig.get(LINE_BREAKER_CONF).length() == 1) {
+            lineBreaker = taskConfig.get(LINE_BREAKER_CONF);
+        } else {
+            lineBreaker = getString(LINE_BREAKER_CONF);
+        }
         maxOutstandingEvents = getInt(MAX_OUTSTANDING_EVENTS_CONF);
         maxRetries = getInt(MAX_RETRIES_CONF);
         backoffThresholdSeconds = getInt(HEC_BACKOFF_PRESSURE_THRESHOLD);

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

@@ -36,6 +36,7 @@ public class ConfigProfile {
     private String headerSource;
     private String headerSourcetype;
     private String headerHost;
+    private String lineBreaker;
 
     public ConfigProfile() {
         this(0);
@@ -88,6 +89,7 @@ public ConfigProfile buildProfileDefault() {
         this.trackData = true;
         this.maxBatchSize = 1;
         this.numOfThreads = 1;
+        this.lineBreaker = "\n";
         return this;
     }
 
@@ -222,6 +224,10 @@ public ConfigProfile buildProfileFour() {
         return this;
     }
 
+    public String getLineBreaker() {
+        return lineBreaker;
+    }
+
     public String getTopics() {
         return topics;
     }

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

@@ -266,6 +266,22 @@ public void testEmptyTopicsAndTopicsRegexCombination() {
         SplunkSinkConnectorConfig connectorConfig = new SplunkSinkConnectorConfig(config);
     }
 
+    @Test
+    public void testSpecialCharLineBreaker() {
+        UnitUtil uu = new UnitUtil(0);
+        Map<String, String> config = uu.createTaskConfig();
+        SplunkSinkConnectorConfig connectorConfig = new SplunkSinkConnectorConfig(config);
+        Assert.assertEquals("\n", connectorConfig.lineBreaker);
+
+        config.put(SplunkSinkConnectorConfig.LINE_BREAKER_CONF, "\r");
+        connectorConfig = new SplunkSinkConnectorConfig(config);
+        Assert.assertEquals("\r", connectorConfig.lineBreaker);
+
+        config.put(SplunkSinkConnectorConfig.LINE_BREAKER_CONF, "\t");
+        connectorConfig = new SplunkSinkConnectorConfig(config);
+        Assert.assertEquals("\t", connectorConfig.lineBreaker);
+    }
+
     @Test
     public void toStr() {
         UnitUtil uu = new UnitUtil(0);

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

@@ -58,6 +58,7 @@ public Map<String, String> createTaskConfig() {
         config.put(SplunkSinkConnectorConfig.TRACK_DATA_CONF, String.valueOf(configProfile.isTrackData()));
         config.put(SplunkSinkConnectorConfig.MAX_BATCH_SIZE_CONF, String.valueOf(configProfile.getMaxBatchSize()));
         config.put(SplunkSinkConnectorConfig.HEC_THREDS_CONF, String.valueOf(configProfile.getNumOfThreads()));
+        config.put(SplunkSinkConnectorConfig.LINE_BREAKER_CONF, configProfile.getLineBreaker());
         return config;
     }