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>

5 files changed, 31 insertions, 9 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
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/config/ApplicationConfigurationTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/config/ApplicationConfigurationTest.java
index 0c56d5af..9ac43426 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/config/ApplicationConfigurationTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/config/ApplicationConfigurationTest.java
@@ -27,6 +27,7 @@ import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -57,6 +58,10 @@ public class ApplicationConfigurationTest {
         assertNotNull(config.getDmaapKafkaGroup());
         assertTrue(config.getDmaapKafkaTimeout() > 0L);
         assertTrue(config.getDmaapCheckNewTopicInterval() > 0);
+        
+        assertNull(config.getDmaapKafkaLogin());
+        assertNull(config.getDmaapKafkaPass());
+        assertNull(config.getDmaapKafkaSecurityProtocol());
 
         assertTrue(config.getKafkaConsumerCount() > 0);
 
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java
index f52332a5..4bc18320 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java
@@ -99,11 +99,14 @@ public class TopicConfigTest {
 
         TopicConfig testTopicConfig = testTopic.getTopicConfig();
         testTopicConfig.setSinkdbs(null);
+        testTopicConfig.setEnabledSinkdbs(null);
         assertFalse(testTopicConfig.supportElasticsearch());
         assertNull(testTopicConfig.getDataFormat2());
                 
         testTopic.setDbs(new HashSet<>());
-        testTopic.getDbs().add(new Db("Elasticsearch"));
+        Db esDb = new Db("Elasticsearch");
+        esDb.setEnabled(true);
+        testTopic.getDbs().add(esDb);
         
         testTopicConfig = testTopic.getTopicConfig();
 
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java
index fc05d1d4..94eeb085 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java
@@ -20,6 +20,7 @@
 
 package org.onap.datalake.feeder.service;
 
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.when;
 
 import java.lang.reflect.InvocationTargetException;
@@ -108,6 +109,12 @@ public class StoreServiceTest {
 		topicConfig.getSinkdbs().add("MongoDB");
 		topicConfig.getSinkdbs().add("HDFS");
 
+
+		topicConfig.setEnabledSinkdbs(new ArrayList<>());
+		topicConfig.getEnabledSinkdbs().add("Elasticsearch");
+		assertTrue(topicConfig.supportElasticsearch());
+		
+		
 		createTopicConfig("test4", "TEXT");
 
 		when(config.getTimestampLabel()).thenReturn("ts");