Check Db enabled flag

If DB is disabled, don't send data

Issue-ID: DCAEGEN2-1622

Change-Id: I24482259125853d28a9abed2b6e987c169411411
Signed-off-by: Guobiao Mo <guobiaomo@chinamobile.com>

2 files changed, 15 insertions, 8 deletions

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
index acb48aef..c171c569 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
@@ -161,14 +161,20 @@ public class Topic {
 		tConfig.setAggregateArrayPath(getAggregateArrayPath());
 		tConfig.setFlattenArrayPath(getFlattenArrayPath());
 		tConfig.setTtl(getTtl());
+		
 		Set<Db> topicDb = getDbs();
 		List<String> dbList = new ArrayList<>();
+		List<String> enabledDbList = new ArrayList<>();
 		if (topicDb != null) {
 			for (Db item : topicDb) {
 				dbList.add(item.getName());
+				if(item.isEnabled()) {
+					enabledDbList.add(item.getName());
+				}
 			}
 		}
 		tConfig.setSinkdbs(dbList);
+		tConfig.setEnabledSinkdbs(enabledDbList);
 
 		return tConfig;
 	}
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
index 8dfe1b16..70778bb3 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
@@ -45,6 +45,7 @@ public class TopicConfig {
 	private String login;
 	private String password;
 	private List<String> sinkdbs;
+	private List<String> enabledSinkdbs;//only include enabled db
 	private boolean enabled;
 	private boolean saveRaw;
 	private String dataFormat;
@@ -63,27 +64,27 @@ public class TopicConfig {
 	}
 
 	public boolean supportHdfs() {
-		return containDb("HDFS");
+		return supportDb("HDFS");
 	}
 
 	public boolean supportElasticsearch() {
-		return containDb("Elasticsearch");//TODO string hard codes
+		return supportDb("Elasticsearch");//TODO string hard codes
 	}
 
 	public boolean supportCouchbase() {
-		return containDb("Couchbase");
+		return supportDb("Couchbase");
 	}
 
 	public boolean supportDruid() {
-		return containDb("Druid");
+		return supportDb("Druid");
 	}
 
 	public boolean supportMongoDB() {
-		return containDb("MongoDB");
+		return supportDb("MongoDB");
 	}
 
-	private boolean containDb(String dbName) {
-		return (sinkdbs != null && sinkdbs.contains(dbName));
+	private boolean supportDb(String dbName) {
+		return (enabledSinkdbs != null && enabledSinkdbs.contains(dbName));
 	}
 
 	//extract DB id from JSON attributes, support multiple attributes
@@ -128,7 +129,7 @@ public class TopicConfig {
 
 	@Override
 	public String toString() {
-		return name;
+		return String.format("Topic %s(enabled=%s, enabledSinkdbs=%s)", name, enabled, enabledSinkdbs);
 	}
 
 	@Override