Unit test code for datalake seed code

Fixed the folder related path for generating
and reading files from correct folder

Issue-ID: DCAEGEN2-1309

Change-Id: I1ec3aae549f484e667eb4048c1ab59b6a60aaf87
Signed-off-by: Rama-Huawei <rama.subba.reddy.s@huawei.com>

6 files changed, 1140 insertions, 144 deletions

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java
index 31f46362..8a177cc7 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java
@@ -1,24 +1,34 @@
 /*

-* ============LICENSE_START=======================================================

-* ONAP : DataLake

-* ================================================================================

-* Copyright 2019 China Mobile

-*=================================================================================

-* Licensed under the Apache License, Version 2.0 (the "License");

-* you may not use this file except in compliance with the License.

-* You may obtain a copy of the License at

-*

-*     http://www.apache.org/licenses/LICENSE-2.0

-*

-* Unless required by applicable law or agreed to in writing, software

-* distributed under the License is distributed on an "AS IS" BASIS,

-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

-* See the License for the specific language governing permissions and

-* limitations under the License.

-* ============LICENSE_END=========================================================

-*/

+ * ============LICENSE_START=======================================================

+ * ONAP : DataLake

+ * ================================================================================

+ * Copyright 2019 China Mobile

+ *=================================================================================

+ * Licensed under the Apache License, Version 2.0 (the "License");

+ * you may not use this file except in compliance with the License.

+ * You may obtain a copy of the License at

+ *

+ *     http://www.apache.org/licenses/LICENSE-2.0

+ *

+ * Unless required by applicable law or agreed to in writing, software

+ * distributed under the License is distributed on an "AS IS" BASIS,

+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

+ * See the License for the specific language governing permissions and

+ * limitations under the License.

+ * ============LICENSE_END=========================================================

+ */

 package org.onap.datalake.feeder.util;

 

+import com.fasterxml.jackson.databind.JsonNode;

+import com.fasterxml.jackson.databind.ObjectMapper;

+import com.fasterxml.jackson.databind.node.JsonNodeType;

+import lombok.Getter;

+import org.apache.velocity.Template;

+import org.apache.velocity.VelocityContext;

+import org.apache.velocity.app.Velocity;

+import org.apache.velocity.runtime.RuntimeConstants;

+import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader;

+

 import java.io.BufferedWriter;

 import java.io.FileWriter;

 import java.io.IOException;

@@ -28,29 +38,15 @@ import java.util.Iterator;
 import java.util.List;

 import java.util.Map.Entry;

 

-import org.apache.velocity.Template;

-import org.apache.velocity.VelocityContext;

-import org.apache.velocity.app.Velocity;

-import org.apache.velocity.runtime.RuntimeConstants;

-import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader;

-import org.onap.datalake.feeder.enumeration.DataFormat;

-

-import com.fasterxml.jackson.databind.JsonNode;

-import com.fasterxml.jackson.databind.ObjectMapper;

-import com.fasterxml.jackson.databind.node.JsonNodeType;

-

-import lombok.Getter;

-import lombok.Setter;

-

 

 /*

  * read sample json and output supervisor to  resources\druid\generated

  * need manual edit to be production ready, final versions are in resources\druid

- * 

+ *

  * http://druid.io/docs/latest/tutorials/tutorial-ingestion-spec.html

  * http://druid.io/docs/latest/ingestion/flatten-json

- * 

- * 

+ *

+ *

  * todo:

  * reduce the manual editing

  * path hard coded

@@ -63,67 +59,67 @@ import lombok.Setter;
 @Getter

 public class DruidSupervisorGenerator {

 

-	Template template = null;

-	VelocityContext context;

+    Template template = null;

+    VelocityContext context;

 

-	List<String[]> dimensions;

+    List<String[]> dimensions;

 

-	public DruidSupervisorGenerator() {

-		dimensions = new ArrayList<>();

+    public DruidSupervisorGenerator() {

+        dimensions = new ArrayList<>();

 

-		Velocity.setProperty(RuntimeConstants.RESOURCE_LOADER, "classpath");

-		Velocity.setProperty("classpath.resource.loader.class", ClasspathResourceLoader.class.getName());

+        Velocity.setProperty(RuntimeConstants.RESOURCE_LOADER, "classpath");

+        Velocity.setProperty("classpath.resource.loader.class", ClasspathResourceLoader.class.getName());

 

-		Velocity.init();

+        Velocity.init();

 

-		context = new VelocityContext();

+        context = new VelocityContext();

 

-		context.put("host", "message-router-kafka:9092");//TODO get from config

+        context.put("host", "message-router-kafka:9092");//TODO get from config

 

-		template = Velocity.getTemplate("druid/kafka-supervisor-template.vm");

-	}

+        template = Velocity.getTemplate("druid/kafka-supervisor-template.vm");

+    }

 

-	private void printNode(String prefix, JsonNode node) {

+    private void printNode(String prefix, JsonNode node) {

 

-		// lets see what type the node is

-		//		System.out.println("NodeType=" + node.getNodeType() + ", isContainerNode=" + node.isContainerNode() + ", " + node); // prints OBJECT

+        // lets see what type the node is

+        //		System.out.println("NodeType=" + node.getNodeType() + ", isContainerNode=" + node.isContainerNode() + ", " + node); // prints OBJECT

 

-		if (node.isContainerNode()) {

+        if (node.isContainerNode()) {

 

-			Iterator<Entry<String, JsonNode>> fields = node.fields();

+            Iterator<Entry<String, JsonNode>> fields = node.fields();

 

-			while (fields.hasNext()) {

-				Entry<String, JsonNode> field = fields.next();

-				//				System.out.println("--------"+field.getKey()+"--------");

-				printNode(prefix + "." + field.getKey(), field.getValue());

-			}

+            while (fields.hasNext()) {

+                Entry<String, JsonNode> field = fields.next();

+                //				System.out.println("--------"+field.getKey()+"--------");

+                printNode(prefix + "." + field.getKey(), field.getValue());

+            }

 

-			if (node.isArray()) {

-				Iterator<JsonNode> elements = node.elements();

-				int i = 0;

-				while (elements.hasNext()) {

-					JsonNode element = elements.next();

-					printNode(prefix + "[" + i + "]", element);

-					i++;

-				}

-			}

+            if (node.isArray()) {

+                Iterator<JsonNode> elements = node.elements();

+                int i = 0;

+                while (elements.hasNext()) {

+                    JsonNode element = elements.next();

+                    printNode(prefix + "[" + i + "]", element);

+                    i++;

+                }

+            }

 

-		} else {

-			printFlattenSpec(node.getNodeType(), prefix);

-		}

+        } else {

+            printFlattenSpec(node.getNodeType(), prefix);

+        }

 

-	}

+    }

 

-	private void printFlattenSpec(JsonNodeType type, String path) {

-		String name = path.substring(2).replace('.', ':');

-		// lets see what type the node is

-		System.out.println("{");

-		System.out.println("\"type\": \"path\",");

-		System.out.println("\"name\": \"" + name + "\",");

-		System.out.println("\"expr\": \"" + path + "\"");

-		System.out.println("},");

+    private void printFlattenSpec(JsonNodeType type, String path) {

+        String name = path.substring(2).replace('.', ':');

+        // lets see what type the node is

+        System.out.println("{");

+        System.out.println("\"type\": \"path\",");

+        System.out.println("\"name\": \"" + name + "\",");

+        System.out.println("\"expr\": \"" + path + "\"");

+        System.out.println("},");

 

-		dimensions.add(new String[] { name, path });

+        dimensions.add(new String[]{name, path});

 		/*

 		 //for  dimensionsSpec

 				if (JsonNodeType.NUMBER.equals(type)) {

@@ -136,41 +132,41 @@ public class DruidSupervisorGenerator {
 		

 				}

 		*/

-	}

+    }

 

-	public void doTopic(String topic) throws IOException {

-		dimensions.clear();

+    public void doTopic(String topic) throws IOException {

+        dimensions.clear();

 

-		String sampleFileName = "C:\\git\\onap\\datalake\\olap\\src\\main\\resources\\druid\\" + topic + "-sample-format.json";//FIXME hard coded path

-		String outputFileName = "C:\\git\\onap\\datalake\\olap\\src\\main\\resources\\druid\\generated\\" + topic + "-kafka-supervisor.json";

+        String sampleFileName = "src/main/resources/druid/" + topic + "-sample-format.json";//FIXME hard coded path

+        String outputFileName = "src/main/resources/druid/generated/" + topic + "-kafka-supervisor.json";

 

-		// Get the contents of json as a string using commons IO IOUTils class.

-		String sampleJson = Util.getTextFromFile(sampleFileName);

+        // Get the contents of json as a string using commons IO IOUTils class.

+        String sampleJson = Util.getTextFromFile(sampleFileName);

 

-		// create an ObjectMapper instance.

-		ObjectMapper mapper = new ObjectMapper();

-		// use the ObjectMapper to read the json string and create a tree

-		JsonNode root = mapper.readTree(sampleJson);

-		printNode("$", root);

+        // create an ObjectMapper instance.

+        ObjectMapper mapper = new ObjectMapper();

+        // use the ObjectMapper to read the json string and create a tree

+        JsonNode root = mapper.readTree(sampleJson);

+        printNode("$", root);

 

-		context.put("topic", topic);

-		context.put("timestamp", "event-header:timestamp");//FIXME hard coded, should be topic based

-		context.put("timestampFormat", "yyyyMMdd-HH:mm:ss:SSS");//FIXME hard coded, should be topic based

-		context.put("dimensions", dimensions);

+        context.put("topic", topic);

+        context.put("timestamp", "event-header:timestamp");//FIXME hard coded, should be topic based

+        context.put("timestampFormat", "yyyyMMdd-HH:mm:ss:SSS");//FIXME hard coded, should be topic based

+        context.put("dimensions", dimensions);

 

-		BufferedWriter out = new BufferedWriter(new FileWriter(outputFileName));

+        BufferedWriter out = new BufferedWriter(new FileWriter(outputFileName));

 

-		template.merge(context, out);

-		out.close();

-	}

+        template.merge(context, out);

+        out.close();

+    }

 

-	public static void main(String[] args) throws MalformedURLException, IOException {

-		String[] topics = new String[] { "AAI-EVENT", "msgrtr.apinode.metrics.dmaap", "unauthenticated.DCAE_CL_OUTPUT", "unauthenticated.SEC_FAULT_OUTPUT" };//FIXME hard coded

+    public static void main(String[] args) throws MalformedURLException, IOException {

+        String[] topics = new String[]{"AAI-EVENT", "msgrtr.apinode.metrics.dmaap", "unauthenticated.DCAE_CL_OUTPUT", "unauthenticated.SEC_FAULT_OUTPUT"};//FIXME hard coded

 

-		DruidSupervisorGenerator p = new DruidSupervisorGenerator();

+        DruidSupervisorGenerator p = new DruidSupervisorGenerator();

 

-		for (String topic : topics) {

-			p.doTopic(topic);

-		}

-	}

+        for (String topic : topics) {

+            p.doTopic(topic);

+        }

+    }

 }

diff --git a/components/datalake-handler/feeder/src/main/resources/druid/generated/AAI-EVENT-kafka-supervisor.json b/components/datalake-handler/feeder/src/main/resources/druid/generated/AAI-EVENT-kafka-supervisor.json
new file mode 100644
index 00000000..72c21342
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/resources/druid/generated/AAI-EVENT-kafka-supervisor.json
@@ -0,0 +1,289 @@
+{
+	"type": "kafka",
+	"dataSchema": {
+		"dataSource": "AAI-EVENT",
+		"parser": {
+			"type": "string",
+			"parseSpec": {
+				"format": "json",
+				"flattenSpec": {
+					"useFieldDiscovery": false,
+					"fields": [
+					
+					{
+"type": "path",
+"name": "_id",
+"expr": "$._id"
+},
+					
+					{
+"type": "path",
+"name": "cambria:partition",
+"expr": "$.cambria.partition"
+},
+					
+					{
+"type": "path",
+"name": "event-header:severity",
+"expr": "$.event-header.severity"
+},
+					
+					{
+"type": "path",
+"name": "event-header:entity-type",
+"expr": "$.event-header.entity-type"
+},
+					
+					{
+"type": "path",
+"name": "event-header:top-entity-type",
+"expr": "$.event-header.top-entity-type"
+},
+					
+					{
+"type": "path",
+"name": "event-header:entity-link",
+"expr": "$.event-header.entity-link"
+},
+					
+					{
+"type": "path",
+"name": "event-header:event-type",
+"expr": "$.event-header.event-type"
+},
+					
+					{
+"type": "path",
+"name": "event-header:domain",
+"expr": "$.event-header.domain"
+},
+					
+					{
+"type": "path",
+"name": "event-header:action",
+"expr": "$.event-header.action"
+},
+					
+					{
+"type": "path",
+"name": "event-header:sequence-number",
+"expr": "$.event-header.sequence-number"
+},
+					
+					{
+"type": "path",
+"name": "event-header:id",
+"expr": "$.event-header.id"
+},
+					
+					{
+"type": "path",
+"name": "event-header:source-name",
+"expr": "$.event-header.source-name"
+},
+					
+					{
+"type": "path",
+"name": "event-header:version",
+"expr": "$.event-header.version"
+},
+					
+					{
+"type": "path",
+"name": "event-header:timestamp",
+"expr": "$.event-header.timestamp"
+},
+					
+					{
+"type": "path",
+"name": "entity:thirdparty-sdnc-id",
+"expr": "$.entity.thirdparty-sdnc-id"
+},
+					
+					{
+"type": "path",
+"name": "entity:resource-version",
+"expr": "$.entity.resource-version"
+},
+					
+					{
+"type": "path",
+"name": "entity:location",
+"expr": "$.entity.location"
+},
+					
+					{
+"type": "path",
+"name": "entity:product-name",
+"expr": "$.entity.product-name"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:esr-system-info-id",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].esr-system-info-id"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:system-type",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].system-type"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:service-url",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].service-url"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:ssl-cacert",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].ssl-cacert"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:type",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].type"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:ssl-insecure",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].ssl-insecure"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:system-status",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].system-status"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:version",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].version"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:passive",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].passive"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:password",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].password"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:protocol",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].protocol"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:ip-address",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].ip-address"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:cloud-domain",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].cloud-domain"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:user-name",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].user-name"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:system-name",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].system-name"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:port",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].port"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:vendor",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].vendor"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:resource-version",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].resource-version"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:remote-path",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].remote-path"
+},
+					
+					{
+"type": "path",
+"name": "entity:esr-system-info-list:esr-system-info[0]:default-tenant",
+"expr": "$.entity.esr-system-info-list.esr-system-info[0].default-tenant"
+},
+					
+					{
+"type": "path",
+"name": "_dl_type_",
+"expr": "$._dl_type_"
+},
+					
+					{
+"type": "path",
+"name": "_dl_text_",
+"expr": "$._dl_text_"
+},
+					]
+				},
+				"timestampSpec": {
+					"column": "event-header:timestamp",
+					"format": "yyyyMMdd-HH:mm:ss:SSS"
+				},
+				"dimensionsSpec": {
+					"dimensions": [
+					],
+					"dimensionsExclusions": [
+					]
+				}
+			}
+		},
+		"metricsSpec": [],
+		"granularitySpec": {
+			"type": "uniform",
+			"segmentGranularity": "HOUR",
+			"queryGranularity": "MINUTE",
+			"rollup": false
+		}
+	},
+	"tuningConfig": {
+		"type": "kafka",
+		"reportParseExceptions": true
+	},
+	"ioConfig": {
+		"topic": "AAI-EVENT",
+		"replicas": 1,
+		"startDelay": "PT1S",
+		"taskDuration": "PT1H",
+		"completionTimeout": "PT30M",
+		"consumerProperties": {
+			"bootstrap.servers": "message-router-kafka:9092"
+		},
+		"useEarliestOffset": true
+	}
+}
diff --git a/components/datalake-handler/feeder/src/main/resources/druid/generated/msgrtr.apinode.metrics.dmaap-kafka-supervisor.json b/components/datalake-handler/feeder/src/main/resources/druid/generated/msgrtr.apinode.metrics.dmaap-kafka-supervisor.json
new file mode 100644
index 00000000..f1779e92
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/resources/druid/generated/msgrtr.apinode.metrics.dmaap-kafka-supervisor.json
@@ -0,0 +1,283 @@
+{
+	"type": "kafka",
+	"dataSchema": {
+		"dataSource": "msgrtr.apinode.metrics.dmaap",
+		"parser": {
+			"type": "string",
+			"parseSpec": {
+				"format": "json",
+				"flattenSpec": {
+					"useFieldDiscovery": false,
+					"fields": [
+					
+					{
+"type": "path",
+"name": "_id:$oid",
+"expr": "$._id.$oid"
+},
+					
+					{
+"type": "path",
+"name": "sendEpsShort:summary",
+"expr": "$.sendEpsShort.summary"
+},
+					
+					{
+"type": "path",
+"name": "sendEpsShort:raw",
+"expr": "$.sendEpsShort.raw"
+},
+					
+					{
+"type": "path",
+"name": "recvEpsInstant:summary",
+"expr": "$.recvEpsInstant.summary"
+},
+					
+					{
+"type": "path",
+"name": "recvEpsInstant:raw",
+"expr": "$.recvEpsInstant.raw"
+},
+					
+					{
+"type": "path",
+"name": "fanOut:summary",
+"expr": "$.fanOut.summary"
+},
+					
+					{
+"type": "path",
+"name": "fanOut:raw",
+"expr": "$.fanOut.raw"
+},
+					
+					{
+"type": "path",
+"name": "sendEpsLong:summary",
+"expr": "$.sendEpsLong.summary"
+},
+					
+					{
+"type": "path",
+"name": "sendEpsLong:raw",
+"expr": "$.sendEpsLong.raw"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerTimeouts:summary",
+"expr": "$.kafkaConsumerTimeouts.summary"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerTimeouts:raw",
+"expr": "$.kafkaConsumerTimeouts.raw"
+},
+					
+					{
+"type": "path",
+"name": "recvEpsLong:summary",
+"expr": "$.recvEpsLong.summary"
+},
+					
+					{
+"type": "path",
+"name": "recvEpsLong:raw",
+"expr": "$.recvEpsLong.raw"
+},
+					
+					{
+"type": "path",
+"name": "sendEpsInstant:summary",
+"expr": "$.sendEpsInstant.summary"
+},
+					
+					{
+"type": "path",
+"name": "sendEpsInstant:raw",
+"expr": "$.sendEpsInstant.raw"
+},
+					
+					{
+"type": "path",
+"name": "recvEpsShort:summary",
+"expr": "$.recvEpsShort.summary"
+},
+					
+					{
+"type": "path",
+"name": "recvEpsShort:raw",
+"expr": "$.recvEpsShort.raw"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerClaims:summary",
+"expr": "$.kafkaConsumerClaims.summary"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerClaims:raw",
+"expr": "$.kafkaConsumerClaims.raw"
+},
+					
+					{
+"type": "path",
+"name": "version:summary",
+"expr": "$.version.summary"
+},
+					
+					{
+"type": "path",
+"name": "version:raw",
+"expr": "$.version.raw"
+},
+					
+					{
+"type": "path",
+"name": "upTime:summary",
+"expr": "$.upTime.summary"
+},
+					
+					{
+"type": "path",
+"name": "upTime:raw",
+"expr": "$.upTime.raw"
+},
+					
+					{
+"type": "path",
+"name": "sendTotalEvents:summary",
+"expr": "$.sendTotalEvents.summary"
+},
+					
+					{
+"type": "path",
+"name": "sendTotalEvents:raw",
+"expr": "$.sendTotalEvents.raw"
+},
+					
+					{
+"type": "path",
+"name": "hostname",
+"expr": "$.hostname"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerCacheMiss:summary",
+"expr": "$.kafkaConsumerCacheMiss.summary"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerCacheMiss:raw",
+"expr": "$.kafkaConsumerCacheMiss.raw"
+},
+					
+					{
+"type": "path",
+"name": "metricsSendTime",
+"expr": "$.metricsSendTime"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerCacheHit:summary",
+"expr": "$.kafkaConsumerCacheHit.summary"
+},
+					
+					{
+"type": "path",
+"name": "kafkaConsumerCacheHit:raw",
+"expr": "$.kafkaConsumerCacheHit.raw"
+},
+					
+					{
+"type": "path",
+"name": "now",
+"expr": "$.now"
+},
+					
+					{
+"type": "path",
+"name": "transactionEnabled",
+"expr": "$.transactionEnabled"
+},
+					
+					{
+"type": "path",
+"name": "startTime:summary",
+"expr": "$.startTime.summary"
+},
+					
+					{
+"type": "path",
+"name": "startTime:raw",
+"expr": "$.startTime.raw"
+},
+					
+					{
+"type": "path",
+"name": "recvTotalEvents:summary",
+"expr": "$.recvTotalEvents.summary"
+},
+					
+					{
+"type": "path",
+"name": "recvTotalEvents:raw",
+"expr": "$.recvTotalEvents.raw"
+},
+					
+					{
+"type": "path",
+"name": "_dl_type_",
+"expr": "$._dl_type_"
+},
+					
+					{
+"type": "path",
+"name": "_dl_text_",
+"expr": "$._dl_text_"
+},
+					]
+				},
+				"timestampSpec": {
+					"column": "event-header:timestamp",
+					"format": "yyyyMMdd-HH:mm:ss:SSS"
+				},
+				"dimensionsSpec": {
+					"dimensions": [
+					],
+					"dimensionsExclusions": [
+					]
+				}
+			}
+		},
+		"metricsSpec": [],
+		"granularitySpec": {
+			"type": "uniform",
+			"segmentGranularity": "HOUR",
+			"queryGranularity": "MINUTE",
+			"rollup": false
+		}
+	},
+	"tuningConfig": {
+		"type": "kafka",
+		"reportParseExceptions": true
+	},
+	"ioConfig": {
+		"topic": "msgrtr.apinode.metrics.dmaap",
+		"replicas": 1,
+		"startDelay": "PT1S",
+		"taskDuration": "PT1H",
+		"completionTimeout": "PT30M",
+		"consumerProperties": {
+			"bootstrap.servers": "message-router-kafka:9092"
+		},
+		"useEarliestOffset": true
+	}
+}
diff --git a/components/datalake-handler/feeder/src/main/resources/druid/generated/unauthenticated.DCAE_CL_OUTPUT-kafka-supervisor.json b/components/datalake-handler/feeder/src/main/resources/druid/generated/unauthenticated.DCAE_CL_OUTPUT-kafka-supervisor.json
new file mode 100644
index 00000000..fb508e8b
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/resources/druid/generated/unauthenticated.DCAE_CL_OUTPUT-kafka-supervisor.json
@@ -0,0 +1,175 @@
+{
+	"type": "kafka",
+	"dataSchema": {
+		"dataSource": "unauthenticated.DCAE_CL_OUTPUT",
+		"parser": {
+			"type": "string",
+			"parseSpec": {
+				"format": "json",
+				"flattenSpec": {
+					"useFieldDiscovery": false,
+					"fields": [
+					
+					{
+"type": "path",
+"name": "_id",
+"expr": "$._id"
+},
+					
+					{
+"type": "path",
+"name": "closedLoopEventClient",
+"expr": "$.closedLoopEventClient"
+},
+					
+					{
+"type": "path",
+"name": "policyVersion",
+"expr": "$.policyVersion"
+},
+					
+					{
+"type": "path",
+"name": "policyName",
+"expr": "$.policyName"
+},
+					
+					{
+"type": "path",
+"name": "policyScope",
+"expr": "$.policyScope"
+},
+					
+					{
+"type": "path",
+"name": "target_type",
+"expr": "$.target_type"
+},
+					
+					{
+"type": "path",
+"name": "AAI:serviceType",
+"expr": "$.AAI.serviceType"
+},
+					
+					{
+"type": "path",
+"name": "AAI:service-instance_service-instance-id",
+"expr": "$.AAI.service-instance_service-instance-id"
+},
+					
+					{
+"type": "path",
+"name": "AAI:globalSubscriberId",
+"expr": "$.AAI.globalSubscriberId"
+},
+					
+					{
+"type": "path",
+"name": "AAI:vserver_vserver-name",
+"expr": "$.AAI.vserver_vserver-name"
+},
+					
+					{
+"type": "path",
+"name": "AAI:network-information_network-id",
+"expr": "$.AAI.network-information_network-id"
+},
+					
+					{
+"type": "path",
+"name": "closedLoopAlarmStart",
+"expr": "$.closedLoopAlarmStart"
+},
+					
+					{
+"type": "path",
+"name": "closedLoopEventStatus",
+"expr": "$.closedLoopEventStatus"
+},
+					
+					{
+"type": "path",
+"name": "version",
+"expr": "$.version"
+},
+					
+					{
+"type": "path",
+"name": "closedLoopControlName",
+"expr": "$.closedLoopControlName"
+},
+					
+					{
+"type": "path",
+"name": "target",
+"expr": "$.target"
+},
+					
+					{
+"type": "path",
+"name": "closedLoopAlarmEnd",
+"expr": "$.closedLoopAlarmEnd"
+},
+					
+					{
+"type": "path",
+"name": "requestID",
+"expr": "$.requestID"
+},
+					
+					{
+"type": "path",
+"name": "from",
+"expr": "$.from"
+},
+					
+					{
+"type": "path",
+"name": "_dl_type_",
+"expr": "$._dl_type_"
+},
+					
+					{
+"type": "path",
+"name": "_dl_text_",
+"expr": "$._dl_text_"
+},
+					]
+				},
+				"timestampSpec": {
+					"column": "event-header:timestamp",
+					"format": "yyyyMMdd-HH:mm:ss:SSS"
+				},
+				"dimensionsSpec": {
+					"dimensions": [
+					],
+					"dimensionsExclusions": [
+					]
+				}
+			}
+		},
+		"metricsSpec": [],
+		"granularitySpec": {
+			"type": "uniform",
+			"segmentGranularity": "HOUR",
+			"queryGranularity": "MINUTE",
+			"rollup": false
+		}
+	},
+	"tuningConfig": {
+		"type": "kafka",
+		"reportParseExceptions": true
+	},
+	"ioConfig": {
+		"topic": "unauthenticated.DCAE_CL_OUTPUT",
+		"replicas": 1,
+		"startDelay": "PT1S",
+		"taskDuration": "PT1H",
+		"completionTimeout": "PT30M",
+		"consumerProperties": {
+			"bootstrap.servers": "message-router-kafka:9092"
+		},
+		"useEarliestOffset": true
+	}
+}
diff --git a/components/datalake-handler/feeder/src/main/resources/druid/generated/unauthenticated.SEC_FAULT_OUTPUT-kafka-supervisor.json b/components/datalake-handler/feeder/src/main/resources/druid/generated/unauthenticated.SEC_FAULT_OUTPUT-kafka-supervisor.json
new file mode 100644
index 00000000..517b199e
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/resources/druid/generated/unauthenticated.SEC_FAULT_OUTPUT-kafka-supervisor.json
@@ -0,0 +1,247 @@
+{
+	"type": "kafka",
+	"dataSchema": {
+		"dataSource": "unauthenticated.SEC_FAULT_OUTPUT",
+		"parser": {
+			"type": "string",
+			"parseSpec": {
+				"format": "json",
+				"flattenSpec": {
+					"useFieldDiscovery": false,
+					"fields": [
+					
+					{
+"type": "path",
+"name": "_id",
+"expr": "$._id"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:sourceId",
+"expr": "$.event.commonEventHeader.sourceId"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:startEpochMicrosec",
+"expr": "$.event.commonEventHeader.startEpochMicrosec"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:eventId",
+"expr": "$.event.commonEventHeader.eventId"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:domain",
+"expr": "$.event.commonEventHeader.domain"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:lastEpochMicrosec",
+"expr": "$.event.commonEventHeader.lastEpochMicrosec"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:eventName",
+"expr": "$.event.commonEventHeader.eventName"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:sourceName",
+"expr": "$.event.commonEventHeader.sourceName"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:priority",
+"expr": "$.event.commonEventHeader.priority"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:version",
+"expr": "$.event.commonEventHeader.version"
+},
+					
+					{
+"type": "path",
+"name": "event:commonEventHeader:reportingEntityName",
+"expr": "$.event.commonEventHeader.reportingEntityName"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:eventSeverity",
+"expr": "$.event.faultFields.eventSeverity"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmCondition",
+"expr": "$.event.faultFields.alarmCondition"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:faultFieldsVersion",
+"expr": "$.event.faultFields.faultFieldsVersion"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:specificProblem",
+"expr": "$.event.faultFields.specificProblem"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[0]:name",
+"expr": "$.event.faultFields.alarmAdditionalInformation[0].name"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[0]:value",
+"expr": "$.event.faultFields.alarmAdditionalInformation[0].value"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[1]:name",
+"expr": "$.event.faultFields.alarmAdditionalInformation[1].name"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[1]:value",
+"expr": "$.event.faultFields.alarmAdditionalInformation[1].value"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[2]:name",
+"expr": "$.event.faultFields.alarmAdditionalInformation[2].name"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[2]:value",
+"expr": "$.event.faultFields.alarmAdditionalInformation[2].value"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[3]:name",
+"expr": "$.event.faultFields.alarmAdditionalInformation[3].name"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[3]:value",
+"expr": "$.event.faultFields.alarmAdditionalInformation[3].value"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[4]:name",
+"expr": "$.event.faultFields.alarmAdditionalInformation[4].name"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[4]:value",
+"expr": "$.event.faultFields.alarmAdditionalInformation[4].value"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[5]:name",
+"expr": "$.event.faultFields.alarmAdditionalInformation[5].name"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[5]:value",
+"expr": "$.event.faultFields.alarmAdditionalInformation[5].value"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[6]:name",
+"expr": "$.event.faultFields.alarmAdditionalInformation[6].name"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:alarmAdditionalInformation[6]:value",
+"expr": "$.event.faultFields.alarmAdditionalInformation[6].value"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:eventSourceType",
+"expr": "$.event.faultFields.eventSourceType"
+},
+					
+					{
+"type": "path",
+"name": "event:faultFields:vfStatus",
+"expr": "$.event.faultFields.vfStatus"
+},
+					
+					{
+"type": "path",
+"name": "_dl_type_",
+"expr": "$._dl_type_"
+},
+					
+					{
+"type": "path",
+"name": "_dl_text_",
+"expr": "$._dl_text_"
+},
+					]
+				},
+				"timestampSpec": {
+					"column": "event-header:timestamp",
+					"format": "yyyyMMdd-HH:mm:ss:SSS"
+				},
+				"dimensionsSpec": {
+					"dimensions": [
+					],
+					"dimensionsExclusions": [
+					]
+				}
+			}
+		},
+		"metricsSpec": [],
+		"granularitySpec": {
+			"type": "uniform",
+			"segmentGranularity": "HOUR",
+			"queryGranularity": "MINUTE",
+			"rollup": false
+		}
+	},
+	"tuningConfig": {
+		"type": "kafka",
+		"reportParseExceptions": true
+	},
+	"ioConfig": {
+		"topic": "unauthenticated.SEC_FAULT_OUTPUT",
+		"replicas": 1,
+		"startDelay": "PT1S",
+		"taskDuration": "PT1H",
+		"completionTimeout": "PT30M",
+		"consumerProperties": {
+			"bootstrap.servers": "message-router-kafka:9092"
+		},
+		"useEarliestOffset": true
+	}
+}
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/DruidSupervisorGeneratorTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/DruidSupervisorGeneratorTest.java
index f5a42bbb..8a9f0779 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/DruidSupervisorGeneratorTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/DruidSupervisorGeneratorTest.java
@@ -1,39 +1,41 @@
 /*

-* ============LICENSE_START=======================================================

-* ONAP : DataLake

-* ================================================================================

-* Copyright 2019 China Mobile

-*=================================================================================

-* Licensed under the Apache License, Version 2.0 (the "License");

-* you may not use this file except in compliance with the License.

-* You may obtain a copy of the License at

-*

-*     http://www.apache.org/licenses/LICENSE-2.0

-*

-* Unless required by applicable law or agreed to in writing, software

-* distributed under the License is distributed on an "AS IS" BASIS,

-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

-* See the License for the specific language governing permissions and

-* limitations under the License.

-* ============LICENSE_END=========================================================

-*/

+ * ============LICENSE_START=======================================================

+ * ONAP : DataLake

+ * ================================================================================

+ * Copyright 2019 China Mobile

+ *=================================================================================

+ * Licensed under the Apache License, Version 2.0 (the "License");

+ * you may not use this file except in compliance with the License.

+ * You may obtain a copy of the License at

+ *

+ *     http://www.apache.org/licenses/LICENSE-2.0

+ *

+ * Unless required by applicable law or agreed to in writing, software

+ * distributed under the License is distributed on an "AS IS" BASIS,

+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

+ * See the License for the specific language governing permissions and

+ * limitations under the License.

+ * ============LICENSE_END=========================================================

+ */

 package org.onap.datalake.feeder.util;

 

-import static org.junit.Assert.assertEquals;

-import static org.junit.Assert.assertNotNull;

-

 import org.apache.velocity.VelocityContext;

 import org.junit.Test;

-import org.junit.runner.RunWith; 

+import org.junit.runner.RunWith;

 import org.onap.datalake.feeder.config.ApplicationConfiguration;

-import org.springframework.beans.factory.annotation.Autowired; 

-import org.springframework.boot.test.context.SpringBootTest; 

+import org.springframework.beans.factory.annotation.Autowired;

+import org.springframework.boot.test.context.SpringBootTest;

 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

 

+import java.io.IOException;

+

+import static org.junit.Assert.assertNotNull;

+import static org.junit.Assert.assertEquals;

+

 /**

  * Test DruidSupervisorGenerator

- * @author Guobiao Mo

  *

+ * @author Guobiao Mo

  */

 

 @RunWith(SpringJUnit4ClassRunner.class)

@@ -41,19 +43,23 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 

 public class DruidSupervisorGeneratorTest {

 

-	@Autowired

-	private ApplicationConfiguration config;

+    @Autowired

+    private ApplicationConfiguration config;

 

     @Test

-	public void testConstructor() {

-		DruidSupervisorGenerator gen = new DruidSupervisorGenerator();

-		VelocityContext context= gen.getContext();

-

-		assertNotNull(context);

-		assertNotNull(gen.getDimensions() );

-		assertNotNull(gen.getTemplate() );

-		

-		String host = (String) context.get("host");		

-		assertEquals(host, config.getDmaapKafkaHostPort());						

-	}

+    public void testConstructor() throws IOException {

+        DruidSupervisorGenerator gen = new DruidSupervisorGenerator();

+        VelocityContext context = gen.getContext();

+

+        assertNotNull(context);

+        assertNotNull(gen.getDimensions());

+        assertNotNull(gen.getTemplate());

+

+        String host = (String) context.get("host");

+        assertEquals(host, config.getDmaapKafkaHostPort());

+

+        String[] strArray2 = {"test1", "test2", "test3"};

+

+        DruidSupervisorGenerator.main(strArray2);

+    }

 }