KYLIN-2775 fix wrong tsColName parsing

Author: yiming187

examples/sample_cube/template/kafka/DEFAULT.KYLIN_STREAMING_TABLE.json

@@ -6,7 +6,7 @@
   "timeout": 60000,
   "bufferSize": 65536,
   "parserName": "org.apache.kylin.source.kafka.TimedJsonStreamParser",
-  "parserProperties": "tsColName=order_time",
+  "parserTimeStampField": "order_time",
   "last_modified": 0,
   "clusters": [
     {

source-kafka/src/main/java/org/apache/kylin/source/kafka/StreamingParser.java

@@ -76,7 +76,7 @@ public static StreamingParser getStreamingParser(String parserName, String parse
             Constructor constructor = clazz.getConstructor(List.class, Map.class);
             return (StreamingParser) constructor.newInstance(columns, properties);
         } else {
-            throw new IllegalStateException("invalid StreamingConfig, parserName " + parserName + ", parserProperties " + parserProperties + ".");
+            throw new IllegalStateException("Invalid StreamingConfig, parserName " + parserName + ", parserProperties " + parserProperties + ".");
         }
     }
 

source-kafka/src/main/java/org/apache/kylin/source/kafka/config/KafkaConfig.java

@@ -34,6 +34,7 @@
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonManagedReference;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.kylin.source.kafka.TimedJsonStreamParser;
 
 /**
  */
@@ -141,12 +142,12 @@ public void setParserProperties(String parserProperties) {
 
     public String getAllParserProperties() {
         StringBuilder sb = new StringBuilder();
-        if (parserProperties != null)
-            sb.append(parserProperties);
         if (parserTimeStampField != null) {
+            sb.append(TimedJsonStreamParser.PROPERTY_TS_COLUMN_NAME + "=" + parserTimeStampField);
             sb.append(";");
-            sb.append(parserTimeStampField);
         }
+        if (parserProperties != null)
+            sb.append(parserProperties);
         return sb.toString();
     }